entity_class: Type[T]) - None:通用方法保存财务报告数据Args:session: 数据库会话report: 报告文件路径code: 公司代码entity_class: 目标实体类(Debt/Cash/Benefit)# 读取原始数据raw_data = read_report(code，后面是程序结束print('程序关闭') 财务分析工具类 [Python] 纯文本查看 复制代码 import loggingfrom decimal import Decimalfrom pathlib import Pathfrom typing import Type，产生一切责任后果自负！ 以上 代码仅用于学习，)# 自动提交， float， index_col=0，可以获取到自增ID，b: Optional[Union[int， format='%Y-%m-%d').date()item.type = 0else:item.year = int(year)item.type = 1debt_list.append(item)return debt_listdef save_financial_report(session: SessionAuto，否则要手动flushclass AutoFlushSession(Session):def add(self， Decimal]， v in data.items()}).encode()except json.JSONDecodeError:passreturn await call_next(request)# 添加请求头及响应头async def auth_token(request: Request， float， v in params.items()}request.scope[query_string] = .join(f{k}={v} for k， Decimal]]]) - Union[int，*， call_next):token = request.headers.get('X-Token')if not token:pass# raise APIException(ErrorCode.UNAUTHORIZED)start_time = time.perf_counter()# 执行当前请求获取响应response: Response = await call_next(request)# 当前响应后执行# time.sleep(1)end_time = (time.perf_counter() - start_time)response.headers[X-Process-Time] = str(end_time)return response# 继承基础类方式实现http拦截器class CustomMiddleware(BaseHTTPMiddleware):async def dispatch(self， request: Request，产生一切责任后果自负！ 以上 代码仅用于学习， float) for num in filtered):return sum(float(num) for num in filtered)return sum(int(num) for num in filtered)def decimal_subtract(a: Optional[Union[int， # commit后触发、query时触发、add/update不会触发expire_on_commit=False， 本帖最后由 a525 于 2025-10-29 11:17 编辑 快速学会fastapi， header=0。
float， PATCH):if application/json in request.headers.get(content-type，report: str。
下载次数: 0) 下载附件 2025-10-27 17:09 上传 HTTP拦截 [Python] 纯文本查看 复制代码 import jsonimport timefrom starlette.middleware.base import BaseHTTPMiddleware， instance， PUT。
'asdfas')return response# 定义类方法(默认第一个参数是当前类本身)返回当前类可以去除pycharm警告# 如果需要给中间件传参， RequestResponseEndpointfrom starlette.requests import Requestfrom starlette.responses import Response'''http 拦截器(中间件)'''# 参数预处理。
Decimal]]，as_percentage: bool = True，strict: bool = False) - Union[int。
data)item.code = codeyear = year.__str__()if len(year) 4:item.quarter = pd.to_datetime(year，ignore_none: bool = True。
股票数据展示基于18步股票分析法 喜欢 请给个start github https://github.com/GET-BUG/fastapi-one-day.git https://github.com/GET-BUG/fastapi-one-day.git https://github.com/GET-BUG/fastapi-one-day.git 项目接口文档地址：:5000/openapi.json 启动方式：直接运行main方法 项目完成功能： 接口统一返回结果 全局异常处理 数据库sqlite集成 数据库事务管理 全局请求响应拦截器 通用restful风格接口实现 前端html集成 同花顺个股网财务报表excel分析 股票财务分析项目实战 image.png (214.38 KB，产生一切责任后果自负！ 以上 代码仅用于学习。
float]]， year_df in df.groupby('Year'):data = year_df.to_dict('records')[0]item = ChineseFieldAdapter.create_from_chinese(cls， Depends(auto_commit_db)]def create_db_and_table():SQLModel.metadata.create_all(engine)print('create tables success.')# 声明程序的生命周期方法， float，entity_class.type， float]:安全计算多个数的差（支持处理 None 值）Args:*numbers: 要计算的数字（可以是 None）default: 当数字为 None 时的默认值（默认为 0）ignore_none: 是否忽略 None 值（True 时替换为 default，适合小白快速上手python， Decimal) for num in filtered):return sum(Decimal(str(num)) for num in filtered)elif any(isinstance(num。
echo=True)# 重写父类方法实现add后立即flush，空字符串转成None防止框架异常(字符串123无法用int接收参数)async def auto_convert_middleware(request: Request， row)) for row in result}# 过滤出需要新增的数据filtered_data = [record for record in raw_dataif f{record.code}-{record.type}-{record.year}-{record.quarter} not in exists]# 批量保存新数据if filtered_data:session.bulk_save_objects(filtered_data)def calculate_ratio(numerator: Union[int， List， skiprows=1)df = df.transpose()df.index.name = Yeardf.reset_index(inplace=True)debt_list: list[T] = []for year，多个要创建成方法塞进来@asynccontextmanagerasync def lifespan(app: FastAPI):print('程序启动')create_db_and_table()yield # yield前是程序启动前， cls: Type[T]) - list[T]:df = pd.read_excel(Path(file_path)，学会fastapi框架完成简单的项目开发， Decimal]， Union， Benefitfrom core.sql_config import SessionAutofrom utils.chinese_to_field_util import ChineseFieldAdapterT = TypeVar('T'， _warn=True):super().add(instance， PUT，default: Optional[Union[int， Benefit)def read_report(code: str， Optionalimport pandas as pdfrom sqlmodel import selectfrom typing_extensions import TypeVarfrom api.entity import Debt。
Debt， connect_args=connect_args，False 时抛出异常）strict: 是否严格检查至少有一个有效数字（True 时若无有效数字则抛出异常）Returns:计算后的差值Raises:ValueError: 如果 ignore_none=False 且遇到 None。
Decimal]] = 0。
file_path: str。
instances: Iterable[object]):super().add_all(instances)self.flush()# 注入自定义逻辑SessionLocal = sessionmaker(bind=engine，allow_negative: bool = True) - Optional[Union[float，default: Union[int。
engine=xlrd， report， float， DELETE，autoflush=True， FastAPIfrom sqlalchemy import create_enginefrom sqlalchemy.orm import sessionmakerfrom sqlmodel import Session，entity_class.quarter).where(entity_class.code == code)).all()# 构建已存在数据的唯一标识集合exists = {-.join(map(str，code: str，。
):body = await request.body()try:data = json.loads(body) if body else {}request._body = json.dumps({k: None if v == else vfor k， call_next: RequestResponseEndpoint):response = await call_next(request)response.headers.append('X-Http'。
POST):params = dict(request.query_params)converted_params = {k: None if v == else v for k，或 strict=True 且全部为 Noneif not numbers:return 0# 处理 strict 模式（无有效数字时报错）if strict and all(x is None for x in numbers):raise ValueError(至少需要一个非 None 的数字)# 处理 ignore_none 逻辑processed_numbers = []for num in numbers:if num is None:if not ignore_none:raise ValueError(不允许 None 值参与计算)processed_numbers.append(default)else:processed_numbers.append(num)# 计算差值（第一个数减去后续所有数的和）if not processed_numbers:return 0return processed_numbers[0] - sum(processed_numbers[1:]) 以上代码仅用于学习。
denominator: Union[int，可以修改方法参数@classmethoddef as_middleware(cls):return cls 数据库集成 [Python] 纯文本查看 复制代码 from contextlib import asynccontextmanagerfrom typing import Annotated， _warn)self.flush()def add_all(self，在启动时创建表# 只能声明一个lifespan， entity_class)# 查询已存在的数据result = session.exec(select(entity_class.code， float] = 0， Decimal]]，precision: int = 4，) - float | None:高精度相减（自动处理None）参数:a: 被减数b: 减数precision: 小数精度位数返回:Decimal(a - b) 或 Nonetry:return (a if a is not None else 0) - (b if b is not None else 0)except:return Nonedef safe_subtract(*numbers: Optional[Union[int， Cash， v in converted_params.items() if v is not None).encode()# 处理请求体 (POST/PUT/PATCH)elif request.method in (POST，产生一切责任后果自负！ ， SQLModelsqlite_file_name = 'database.db'sqlite_url = f'sqlite:///{sqlite_file_name}'connect_args = {check_same_thread: False}engine = create_engine(sqlite_url。
class_=AutoFlushSession， Decimal]:混合类型安全求和参数:numbers: 可能包含int/float/Decimal/None的列表返回:非None值的和（自动提升为最高精度类型）filtered = [num for num in numbers if num is not None]if not filtered:return 0# 自动类型提升逻辑if any(isinstance(num，异常会回滚def auto_commit_db():db = SessionLocal()try:yield dbdb.commit()except Exception:db.rollback()raisefinally:db.close()# Annotated 描述该变量是Session类型# Depends 表示该变量数据源是get_session的返回值SessionAuto = Annotated[Session。
float， call_next):# 处理查询参数 (GET/DELETE)if request.method in (GET，entity_class.year， Decimal]]:通用比例计算方法（修复版）修复点：1. 使用 Decimal 的 quantize() 替代 round()2. 更严格的异常捕获3. 改进的日志记录try:# 转换为Decimal保证计算精度num = Decimal(str(numerator))den = Decimal(str(denominator))# 检查负值if not allow_negative and (num 0 or den 0):return default# 核心计算逻辑result = (num / den) * (100 if as_percentage else 1)# 精度处理if precision 0:result = result.quantize(Decimal(f0.{'0' * precision}))else:result = result.to_integral_value()# 类型转换return float(result) if isinstance(result， Iterablefrom fastapi import Depends， Cash， Decimal) else resultexcept (Exception) as e:logging.error(f'数学计算错误: {numerator}/{denominator} | 错误: {str(e)}')return defaultdef mixed_type_sum(numbers: List[Optional[Union[int， float。