refactor: establish standalone application boundary
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from typing import Any
|
||||
|
||||
|
||||
TUSHARE_URL = "http://api.tushare.pro"
|
||||
|
||||
|
||||
class TushareError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class TushareTransportMixin:
|
||||
def query(
|
||||
self,
|
||||
api_name: str,
|
||||
params: dict[str, Any] | None = None,
|
||||
fields: str = "",
|
||||
) -> list[dict[str, Any]]:
|
||||
payload = json.dumps(
|
||||
{
|
||||
"api_name": api_name,
|
||||
"token": self.token,
|
||||
"params": params or {},
|
||||
"fields": fields,
|
||||
}
|
||||
).encode("utf-8")
|
||||
request = urllib.request.Request(
|
||||
TUSHARE_URL,
|
||||
data=payload,
|
||||
headers={"Content-Type": "application/json", "User-Agent": "XiaobaiReviewWeb/0.2"},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=self.timeout) as response:
|
||||
result = json.loads(response.read().decode("utf-8"))
|
||||
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
|
||||
raise TushareError(f"Tushare request failed: {exc}") from exc
|
||||
|
||||
if result.get("code") != 0:
|
||||
raise TushareError(result.get("msg") or "Tushare returned an unknown error")
|
||||
|
||||
data = result.get("data") or {}
|
||||
columns = data.get("fields") or []
|
||||
return [dict(zip(columns, item)) for item in data.get("items") or []]
|
||||
Reference in New Issue
Block a user