cqxqgtech/wance_data
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重写封装类，规范路由以符合restful风格

Browse Source
This commit is contained in:
mxwj 2024-11-08 15:26:17 +08:00
parent 0a749ff31a
commit f58831a6d3
6 changed files with 87 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
src/__pycache__/responses.cpython-310.pyc
View File

Binary file not shown.

64
src/backtest/router.py
View File

@ -1,32 +1,32 @@
import json
from fastapi import APIRouter
from src.backtest.service import start_backtest_service, stock_chart_service
from src.pydantic.backtest_request import BackRequest
router = APIRouter() # 创建一个 FastAPI 路由器实例
@router.get("/start_backtest")
async def start_backtest(request: BackRequest):
result = await start_backtest_service(field_list=['close', 'time'],
stock_list=request.stock_list,
period=request.period,
start_time=request.start_time,
end_time=request.end_time,
count=request.count,
dividend_type=request.dividend_type,
fill_data=request.fill_data,
ma_type=request.ma_type,
short_window=request.short_window,
long_window=request.long_window
)
return result
@router.get('/stock_chart')
async def stock_chart(request: BackRequest):
result = await stock_chart_service(stock_code=request.stock_code,
benchmark_code=request.benchmark_code)
return result
import json
from fastapi import APIRouter
from src.backtest.service import start_backtest_service, stock_chart_service
from src.pydantic.backtest_request import BackRequest
router = APIRouter() # 创建一个 FastAPI 路由器实例
@router.get("/start_backtest")
async def start_backtest(request: BackRequest):
result = await start_backtest_service(field_list=['close', 'time'],
stock_list=request.stock_list,
period=request.period,
start_time=request.start_time,
end_time=request.end_time,
count=request.count,
dividend_type=request.dividend_type,
fill_data=request.fill_data,
ma_type=request.ma_type,
short_window=request.short_window,
long_window=request.long_window
)
return result
@router.get('/stock_chart')
async def stock_chart(request: BackRequest):
result = await stock_chart_service(stock_code=request.stock_code,
benchmark_code=request.benchmark_code)
return result

6
src/financial_reports/router.py
View File

@ -3,6 +3,7 @@ from fastapi.responses import JSONResponse
from src.financial_reports.schemas import *
from src.financial_reports.service import *
from src.responses import response_list_response
financial_reports_router = APIRouter()
@ -15,11 +16,12 @@ async def financial_repoets_query(request: FinancialReportQuery )-> JSONResponse
return result
@financial_reports_router.post("/keyword_search")
@financial_reports_router.post("/keyword-search")
async def financial_repoets_query_keyword_search(query: SearchKeywordQuery) -> list:
"""
模糊查询接口
"""
result = await user_keyword_search_service(query.searchKeyword)
return result
# 使用封装类返回列表响应
return response_list_response(data=result, message="联想词查询成功")

4
src/models/stock_hu_shen_jing_a.py
View File

@ -1,5 +1,9 @@
from tortoise import fields, Model
"""
待补全每日更新表的爬虫
"""
class StockHuShenJingA(Model):
"""
沪深京 A 股实时行情数据

114
src/responses.py
View File

@ -1,68 +1,46 @@
import typing
from typing import Generic, Sequence, TypeVar
from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from pydantic import BaseModel
T = TypeVar("T")
class EntityResponse(BaseModel, Generic[T]):
"""实体数据"""
status_code: int = 200
message: str = "Success"
data: T
class ListResponse(BaseModel, Generic[T]):
"""列表数据"""
status_code: int = 200
message: str = "Success"
data: Sequence[T]
# 包装响应的 Pydantic 模型
# class EntityPageResponse(BaseModel, Generic[T]):
# status_code: int
# message: str
# data: PageData
def response_entity_response(data, status_code=200, message="Success") -> EntityResponse:
"""普通实体类"""
return EntityResponse(data=data, status_code=status_code, message=message)
# def response_page_response(data, status_code=200, message="Success") -> EntityPageResponse:
# """普通分页类"""
# return EntityPageResponse(data=data, status_code=status_code, message=message)
def response_list_response(data, status_code=200, message="Success") -> ListResponse:
"""普通列表数据"""
return ListResponse(data=data, status_code=status_code, message=message)
def response_success(message: str = 'Success',
headers: typing.Optional[typing.Dict[str, str]] = None) -> JSONResponse:
"""成功返回"""
return JSONResponse(
status_code=status.HTTP_200_OK,
headers=headers,
content=jsonable_encoder({
"status_code": status.HTTP_200_OK,
"message": message,
}))
def response_unauthorized() -> JSONResponse:
"""未登录"""
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
headers={"WWW-Authenticate": "Bearer"},
content=jsonable_encoder({
"status_code": status.HTTP_401_UNAUTHORIZED,
"message": '用户认证失败',
}))
import typing
from typing import Generic, Sequence, TypeVar
from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from pydantic import BaseModel
T = TypeVar("T")
class BaseResponse(BaseModel, Generic[T]):
"""通用响应类"""
status_code: int
message: str
data: typing.Optional[T] = None
def json_response(self) -> JSONResponse:
"""返回 JSONResponse 响应"""
return JSONResponse(
status_code=self.status_code,
content=jsonable_encoder(self.dict())
)
def create_response(data: T = None, status_code: int = 200, message: str = "Success") -> JSONResponse:
"""通用响应函数"""
response = BaseResponse(status_code=status_code, message=message, data=data)
return response.json_response()
# 使用示例
def response_success(message: str = 'Success') -> JSONResponse:
return create_response(status_code=status.HTTP_200_OK, message=message)
def response_unauthorized(message: str = "用户认证失败") -> JSONResponse:
return create_response(status_code=status.HTTP_401_UNAUTHORIZED, message=message)
def response_entity_response(data: T, message: str = "Success") -> JSONResponse:
"""单个实体数据响应"""
return create_response(data=data, status_code=200, message=message)
def response_list_response(data: Sequence[T], message: str = "Success") -> JSONResponse:
"""列表数据响应"""
return create_response(data=data, status_code=200, message=message)

2
src/utils/generate_pinyin_abbreviation.py
View File

@ -3,7 +3,7 @@ from src.models.stock import Stock
async def generate_pinyin_abbreviation():
"""
更新数据库中所有股票记录的拼音缩写，如果字段为空则生成并保存。
更新数据库中所有股票记录的拼音缩写，如果字段为空则生成并保存。,待抽象为通用组件
"""
stocks = await Stock.filter(stock_pinyin__isnull=True) # 查找拼音字段为空的记录
for stock in stocks: