项目框架的搭建

1.项目结构规划1.1主文件main.py 是FastAPI应用主入口from contextlib import asynccontextmanager from fastapi import FastAPI from starlette.middleware.cors import CORSMiddleware from starlette.middleware.trustedhost import TrustedHostMiddleware from tortoise import Tortoise from app.config.settings import settings from app.core.database import TORTOISE_ORM from app.core.exception_handler import global_exception_handler from app.core.logging import logger from app.core.middlewares import LoggingMiddleware # https://fastapi.tiangolo.com/advanced/events/#async-context-manager asynccontextmanager async def lifespan(app: FastAPI): logger.info(正要启动数据库连接) # Tortoise 1.1 uses TortoiseContext (contextvars). Lifespan and request # handlers run in different tasks, so enable global fallback for FastAPI. await Tortoise.init(configTORTOISE_ORM, _enable_global_fallbackTrue) logger.debug(数据库连接启动成功) yield await Tortoise.close_connections() logger.error(数据库连接已关闭) app FastAPI( titlesettings.app_title, versionsettings.app_version, descriptionsettings.app_description, lifespanlifespan ) # 允许跨域 app.add_middleware(CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], allow_credentialsTrue) # 允许 Host访问 app.add_middleware(TrustedHostMiddleware, allowed_hosts[*]) app.add_middleware(LoggingMiddleware) # 全局异常处理 app.add_exception_handler( Exception, global_exception_handler )1.2requirements.txt 存放Python依赖包执行收集所有包pip freeze requirements.txt执行下载文件中所有依赖pip install -r requirements.txt1.3 app 应用核心代码1models # 数据模型层 (Tortoise ORM) 创建表 # 用户相关模型(2) apis # API路由层[接受参数,返回数据] # 用户相关API(3) schems # 数据验证层 (Pydantic) # 用户请求/响应模型(4) services # 业务服务层 (逻辑) # 用户相关的业务代码/方法(5) core # 核心文件(5.1) database.py # 数据库连接信息# app/core/database.py 数据库配置文件 这个文件定义了 Tortoise-ORM 连接 MySQL 数据库所需的所有配置信息 from app.config.settings import settings # TORTOISE_ORM 是 Tortoise-ORM 规定的配置字典变量名 # 后面用 register_tortoise 或 Aerich 时都会引用这个字典 TORTOISE_ORM { # 1. 连接配置 —— 定义数据库连接信息 connections: { # default 是默认连接的名字必须有一个 default default: { # engine指定数据库后端引擎MySQL 使用 tortoise.backends.mysql engine: tortoise.backends.mysql, # credentials数据库连接凭证包含主机、端口、用户名、密码等 credentials: { host: settings.db_host, # MySQL 服务器地址 port: settings.db_port, # MySQL 端口默认 3306 user: settings.db_user, # 数据库用户名 password: settings.db_password, # 数据库密码请根据实际情况修改 database: settings.db_name, # 数据库名称 minsize: settings.db_minsize, # 连接池最小连接数 maxsize: settings.db_maxsize, # 连接池最大连接数 charset: utf8mb4, # 字符集支持 emoji echo: settings.db_echo # 是否打印 SQL 语句开发环境建议开启 } } }, # 2. 应用配置 —— 指定模型所在的模块 apps: { # models 是应用的名字可以自定义但 Aerich 需要使用这个名字 models: { # models 列表指定包含 Tortoise 模型类的 Python 模块路径 # aerich.models 是 Aerich 的内置模型用于记录迁移历史必须包含 models: [app.models, aerich.models], # default_connection指定这个应用使用哪个数据库连接 default_connection: default, } }, # 3. 时区配置 use_tz: False, # 是否使用时区 timezone: Asia/Shanghai, # 时区设置 echo: True # ✅ 关键打开 SQL 打印 }(5.2) logging.py # 日志依赖包pip install logurushell # app/core/logging.py from loguru import logger import sys import logging from pathlib import Path from app.config.settings import settings # 移除默认处理器 logger.remove() # 控制台输出开发环境 logger.add( sys.stdout, formatgreen{time:YYYY-MM-DD HH:mm:ss}/green | level{level: 8}/level | cyan{name}/cyan:cyan{function}/cyan:cyan{line}/cyan - level{message}/level, levelDEBUG if settings.DEBUG else INFO, colorizeTrue ) # 创建日志目录 log_dir Path(logs) log_dir.mkdir(exist_okTrue) # INFO 级别及以上日志文件 logger.add( log_dir / info_{time:YYYY-MM-DD}.log, rotation00:00, # 每天午夜轮转 retention30 days, # 保留 30 天 compressionzip, # 压缩旧日志 levelINFO, format{time:YYYY-MM-DD HH:mm:ss} | {level: 8} | {name}:{function}:{line} - {message}, encodingutf-8, filterlambda record: record[level].name INFO ) # WARNING 级别日志文件 logger.add( log_dir / warning_{time:YYYY-MM-DD}.log, rotation00:00, # 每天午夜轮转 retention30 days, # 保留 30 天 compressionzip, # 压缩旧日志 levelWARNING, format{time:YYYY-MM-DD HH:mm:ss} | {level: 8} | {name}:{function}:{line} - {message}, encodingutf-8, filterlambda record: record[level].name WARNING ) # ERROR 级别及以上日志文件 logger.add( log_dir / error_{time:YYYY-MM-DD}.log, rotation00:00, # 每天午夜轮转 retention90 days, # 保留 90 天错误日志保留更长时间 compressionzip, # 压缩旧日志 levelERROR, format{time:YYYY-MM-DD HH:mm:ss} | {level: 8} | {name}:{function}:{line} - {message}, encodingutf-8, filterlambda record: record[level].name in [ERROR, CRITICAL] ) # 全量日志文件包含所有级别 logger.add( log_dir / all_{time:YYYY-MM-DD}.log, rotation00:00, # 每天午夜轮转 retention7 days, # 保留 7 天 compressionzip, # 压缩旧日志 levelDEBUG, format{time:YYYY-MM-DD HH:mm:ss} | {level: 8} | {name}:{function}:{line} - {message}, encodingutf-8 ) # SQL 日志文件专门记录数据库 SQL 语句 def sql_filter(record): 过滤 SQL 相关的日志 name record[name].lower() message record[message].lower() # 捕获 tortoise 后端相关的日志特别是包含 SQL 语句的日志 return ( tortoise.backends in name or sql in message or select in message or insert in message or update in message or delete in message or create in message or alter in message ) logger.add( log_dir / sql_{time:YYYY-MM-DD}.log, rotation00:00, # 每天午夜轮转 retention30 days, # 保留 30 天 compressionzip, # 压缩旧日志 levelDEBUG, format{time:YYYY-MM-DD HH:mm:ss} | {level: 8} | {name} - {message}, encodingutf-8, filtersql_filter ) # 配置标准 logging 模块将 Tortoise ORM 的日志转发到 loguru class InterceptHandler(logging.Handler): 拦截标准 logging 的输出转发到 loguru def emit(self, record): # 获取对应的 loguru 级别 try: level logger.level(record.levelname).name except ValueError: level record.levelno # 找到调用者信息 frame, depth sys._getframe(6), 6 while frame and frame.f_code.co_filename logging.__file__: frame frame.f_back depth 1 logger.opt(depthdepth, exceptionrecord.exc_info).log(level, record.getMessage()) # 配置 Tortoise ORM 的 logger def setup_tortoise_logging(): 配置 Tortoise ORM 的日志输出 # 拦截所有 tortoise 相关的 logger logging_loggers [ asyncmy, tortoise, tortoise.backends, tortoise.backends.mysql, tortoise.backends.asyncpg, tortoise.backends.sqlite, ] for logger_name in logging_loggers: logging_logger logging.getLogger(logger_name) logging_logger.handlers [InterceptHandler()] logging_logger.setLevel(logging.DEBUG if settings.db_echo else logging.INFO) logging_logger.propagate False # 初始化 Tortoise 日志配置 if settings.db_echo: setup_tortoise_logging() # 导出 logger __all__ [logger, setup_tortoise_logging] (5.3) exception_handler.py # 全局异常处理 全局异常处理 from starlette.requests import Request from starlette.responses import JSONResponse from app.core.logging import logger def global_exception_handler(request: Request, exc: Exception) - JSONResponse: logger.error(出现了Exception异常) return JSONResponse( status_code500, content{ code: 0, message: str(exc) } )(5.4) middlewares.py # 中间件from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint from starlette.requests import Request from starlette.responses import Response from app.core.logging import logger class LoggingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) - Response: method request.method qp request.query_params url request.url logger.info(f{method} {url} {qp}) response await call_next(request) return response(6) config # 配置文件(6.1) settings.py # 多环境配置import os from typing import Optional from pydantic_settings import BaseSettings, SettingsConfigDict class BaseAppSettings(BaseSettings): 基础配置类所有环境共享 model_config SettingsConfigDict( env_file_encodingutf-8, case_sensitiveFalse, extraignore, # 让子类继承 env 配置不会被覆盖掉这是核心修复 env_file.env, ) # 通用配置 app_title: str BOSS服务端项目 app_version: str V1.0.0 api_prefix: str /api/v1 app_description: str Boss项目的接口文档,包含求职者端,企业端,管理端 class DevAppSettings(BaseAppSettings): 开发环境 model_config SettingsConfigDict(env_file.env.dev) # 服务 server_port: int 8000 debug_mode: bool True # 数据库 db_url: str mysqlpymysql://root:123456localhost:3306/fastApiProject004 db_host: str localhost db_port: int 3306 db_user: str root db_password: str root db_name: str boss_api db_echo: bool True db_minsize: int 1 db_maxsize: int 5 DEBUG: bool True test_abc:str 123 # 安全 secret_key: str dev-secret-key-123456-pydantic jwt_token_secret_key: str dhsjjdkfjdkfrjfrjgr-278783jkdsdhjdhjsds-dsdksjdkajskajieuiwueiwhdshmxzxno9iy token_expire_minutes: int 120 cors_allow_origins: list[str] [*] # 阿里云OSS ALIYUN_OSS_ACCESS_KEY_ID: str KEY ALIYUN_OSS_ACCESS_KEY_SECRET: str KEY ALIYUN_OSS_ENDPOINT: str oss-cn-beijing.aliyuncs.com ALIYUN_OSS_BUCKET_NAME: str fastapi-project-004 # 钉钉 DINGTALK_APP_KEY: str bf8ad584-a6d1-454d-8291-5e0158b4722b DINGTALK_REDIRECT_URI: str http://127.0.0.1:8000/third_party/dingtalk/login/callback DINGTALK_CLIENT_ID: str DINGTALK_CLIENT_SECRET: str REDIS_HOST: str 127.0.0.1 REDIS_PORT: int 6379 REDIS_DB: int 8 DEFAULT_AVATAR: str https://img10.360buyimg.com/pcpubliccms/s1440x1440_jfs/t1/240214/16/3793/62089/65acb64bF35c090ae/4cce5ee81fae5a23.jpg.avif # 高德地图 AMAP_SERVRER_KEY: str d0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0 # 微信支付 V3 MCH_ID: str 1558950191 MCH_SERIAL_NO: str 34345964330B66427E0D3D28826C4993C77E631F PRIVATE_KEY_PATH: str apiclient_key.pem API_V3_KEY: str UDuLFDcmy5Eb6o0nTNZdu6ek4DDh4K8B APP_ID: str wx74862e0dfcf69954 DOMAIN: str https://api.mch.weixin.qq.com NOTIFY_DOMAIN: str https://xxx.ngrok.io PARTNER_KEY: str T6m9iK73b0kn9g5v426MKfHQH7X8rKwb property def NATIVE_ORDER_URL(self): return f{self.DOMAIN}/v3/pay/transactions/native property def QUERY_ORDER_URL(self): return f{self.DOMAIN}/v3/pay/transactions/id/ property def CLOSE_ORDER_URL(self): return f{self.DOMAIN}/v3/pay/transactions/out-trade-no/%s/close property def REFUND_URL(self): return f{self.DOMAIN}/v3/refund/domestic/refunds property def QUERY_REFUND_URL(self): return f{self.DOMAIN}/v3/refund/domestic/refunds/%s class TestAppSettings(BaseAppSettings): 测试环境 model_config SettingsConfigDict(env_file.env.test) server_port: int 8001 debug_mode: bool False db_url: str mysqlpymysql://root:123456localhost:3306/test_db secret_key: str test-secret-key-789012-pydantic cors_allow_origins: list[str] [https://test.yourdomain.com] class ProdAppSettings(BaseAppSettings): 生产环境 敏感配置 无默认值必须从环境变量或 .env.prod 读取 更安全、更规范 model_config SettingsConfigDict(env_file.env.prod) server_port: int 80 debug_mode: bool False cors_allow_origins: list[str] [https://yourdomain.com] # 生产必须配置不能为空 db_url: str secret_key: str jwt_token_secret_key: str db_host: str db_port: int db_user: str db_password: str db_name: str # 支付/OSS/第三方 全部从环境变量读取不写死代码 ALIYUN_OSS_ACCESS_KEY_ID: str ALIYUN_OSS_ACCESS_KEY_SECRET: str API_V3_KEY: str MCH_ID: str APP_ID: str # 环境枚举 SUPPORTED_ENVS [dev, test, prod] def get_app_settings(env: Optional[str] None) - BaseAppSettings: 多环境配置工厂标准写法 优先级传入参数 系统环境变量 默认 dev env env or os.getenv(FASTAPI_ENV, dev) if env not in SUPPORTED_ENVS: raise ValueError(f环境错误支持{SUPPORTED_ENVS}) env_map { dev: DevAppSettings, test: TestAppSettings, prod: ProdAppSettings, } return env_map[env]() # 全局唯一配置实例 settings get_app_settings()2.多环境配置一、什么是多环境配置依赖包pip install pydantic-settings多环境配置同一套业务代码在不同运行环境中加载不同的配置值。隔离不同场景的数据库、接口地址、密钥、开关等参数避免改代码切换环境。 常见环境dev开发环境本地开发、联调test测试环境测试人员验收pre预发环境线上镜像灰度验证prod生产环境正式对外服务