| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340 |
- import os
- from functools import lru_cache
- from pathlib import Path
- from typing import Any, Literal
- from urllib.parse import quote_plus
- from pydantic_settings import BaseSettings, SettingsConfigDict
- from app.common.enums import EnvironmentEnum
- from app.config.path_conf import BASE_DIR, ENV_DIR
- class Settings(BaseSettings):
- """系统配置类"""
- model_config = SettingsConfigDict(
- env_file=ENV_DIR / f".env.{os.getenv('ENVIRONMENT')}",
- env_file_encoding="utf-8",
- extra="ignore",
- case_sensitive=True, # 区分大小写
- )
- # ================================================= #
- # ******************* 项目环境 ****************** #
- # ================================================= #
- ENVIRONMENT: EnvironmentEnum = EnvironmentEnum.DEV
- # ================================================= #
- # ******************* 服务器配置 ****************** #
- # ================================================= #
- SERVER_HOST: str = "0.0.0.0" # 允许访问的IP地址
- SERVER_PORT: int = 8001 # 服务端口
- # ================================================= #
- # ******************* API文档配置 ****************** #
- # ================================================= #
- DEBUG: bool = True # 调试模式
- TITLE: str = "🎉 Payment Platform 🎉 " # 文档标题
- VERSION: str = "1.0.0" # 版本号
- DESCRIPTION: str = (
- "Payment Platform API文档" # 文档描述
- )
- SUMMARY: str = "接口汇总" # 文档概述
- DOCS_URL: str = "/docs" # Swagger UI路径
- REDOC_URL: str = "/redoc" # ReDoc路径
- LJDOC_URL: str = "/ljdoc" # LangJin UI路径
- ROOT_PATH: str = "/api/v1" # API路由前缀
- # ================================================= #
- # ******************** 日志配置 ******************** #
- # ================================================= #
- LOGGER_LEVEL: str = "DEBUG" # 日志级别
- # ================================================= #
- # ******************** 跨域配置 ******************** #
- # ================================================= #
- CORS_ORIGIN_ENABLE: bool = True # 是否启用跨域
- ALLOW_ORIGINS: list[str] = ["*"] # 允许的域名列表
- ALLOW_METHODS: list[str] = ["*"] # 允许的HTTP方法
- ALLOW_HEADERS: list[str] = ["*"] # 允许的请求头
- ALLOW_CREDENTIALS: bool = True # 是否允许携带cookie
- CORS_EXPOSE_HEADERS: list[str] = ["X-Request-ID"]
- # ================================================= #
- # ******************* 登录认证配置 ****************** #
- # ================================================= #
- SECRET_KEY: str = "vgb0tnl9d58+6n-6h-ea&u^1#s0ccp!794=krylxcjq75vzps$" # JWT密钥
- ALGORITHM: str = "HS256" # JWT算法
- ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 60 # access_token过期时间(秒)30 分钟
- REFRESH_TOKEN_EXPIRE_MINUTES: int = 60 * 120 # refresh_token过期时间(秒)30 分钟
- TOKEN_TYPE: str = "bearer" # token类型
- TOKEN_REQUEST_PATH_EXCLUDE: list[str] = ["api/v1/auth/login"] # JWT / RBAC 路由白名单
- TOKEN_SLIDING_EXPIRE: bool = True # 是否启用滑动过期(用户操作时自动续期)
- # 多租户:通配子域与登录租户一致(默认关闭;生产按 base 解析 {code}.base)
- TENANT_HOST_ENFORCE: bool = False
- TENANT_HOST_BASE_DOMAIN: str = ""
- TENANT_HOST_IGNORE_PREFIXES: list[str] = ["www", "api", "admin"]
- # ================================================= #
- # ******************** 数据库配置 ******************* #
- # ================================================= #
- SQL_DB_ENABLE: bool = True # 是否启用数据库
- DATABASE_ECHO: bool | Literal["debug"] = False # 是否显示SQL日志
- ECHO_POOL: bool | Literal["debug"] = False # 是否显示连接池日志
- POOL_SIZE: int = 10 # 连接池大小
- MAX_OVERFLOW: int = 20 # 最大溢出连接数
- POOL_TIMEOUT: int = 30 # 连接超时时间(秒)
- POOL_RECYCLE: int = 1800 # 连接回收时间(秒)
- POOL_USE_LIFO: bool = True # 是否使用LIFO连接池
- POOL_PRE_PING: bool = True # 是否开启连接预检
- FUTURE: bool = True # 是否使用SQLAlchemy 2.0特性
- AUTOCOMMIT: bool = False # 是否自动提交
- AUTOFETCH: bool = False # 是否自动刷新
- EXPIRE_ON_COMMIT: bool = False # 是否在提交时过期
- # MySQL/PostgreSQL数据库连接
- DATABASE_TYPE: Literal["mysql", "postgres", "sqlite"] = "mysql"
- DATABASE_HOST: str = "localhost"
- DATABASE_PORT: int = 3306
- DATABASE_USER: str = "root"
- DATABASE_PASSWORD: str = "ServBay.dev"
- DATABASE_NAME: str = "fastapiadmin"
- # ================================================= #
- # ******************** Redis配置 ******************* #
- # ================================================= #
- REDIS_ENABLE: bool = True # 是否启用Redis
- REDIS_HOST: str = "localhost"
- REDIS_PORT: int = 6379
- REDIS_DB_NAME: int = 1
- REDIS_USER: str = ""
- REDIS_PASSWORD: str = ""
- # ================================================= #
- # ******************** 验证码配置 ******************* #
- # ================================================= #
- CAPTCHA_ENABLE: bool = True # 是否启用验证码
- CAPTCHA_EXPIRE_SECONDS: int = 60 * 1 # 验证码过期时间(秒) 1分钟
- CAPTCHA_FONT_SIZE: int = 32 # 字体大小
- CAPTCHA_FONT_PATH: str = "static/assets/font/Arial.ttf" # 字体路径
- # 是否请求外网解析 IP 归属地(登录发 token、操作日志写 login_location 共用;关闭可明显加快登录)
- LOGIN_RESOLVE_IP_LOCATION: bool = False
- # ================================================= #
- # ******************* 外部 HTTP(httpx)******************* #
- # ================================================= #
- HTTPX_DEFAULT_TIMEOUT: float = 10.0 # 对外请求默认超时(秒),见 app/common/httpx_defaults.py
- # ================================================= #
- # ********************* 日志配置 ******************* #
- # ================================================= #
- # 是否额外写入 JSON Lines(loguru ``serialize=True``,与控制台/info.log 同源,便于日志平台采集)
- LOG_JSON_FILE_ENABLE: bool = False
- LOG_JSON_FILE_NAME: str = "app.jsonl" # 相对 LOG_DIR
- LOG_JSON_RETENTION_DAYS: int = 7 # JSON 文件保留天数(通常比文本日志短)
- OPERATION_LOG_RECORD: bool = True # 是否记录操作日志
- IGNORE_OPERATION_FUNCTION: list[str] = ["get_captcha_for_login"] # 忽略记录的函数
- OPERATION_RECORD_METHOD: list[str] = [
- "POST",
- "PUT",
- "PATCH",
- "DELETE",
- "HEAD",
- "OPTIONS",
- ] # 需要记录的请求方法
- # ================================================= #
- # ******************* Gzip压缩配置 ******************* #
- # ================================================= #
- GZIP_ENABLE: bool = True # 是否启用Gzip
- GZIP_MIN_SIZE: int = 1000 # 最小压缩大小(字节)
- GZIP_COMPRESS_LEVEL: int = 9 # 压缩级别(1-9)
- # ================================================= #
- # ***************** 静态文件配置 ***************** #
- # ================================================= #
- STATIC_ENABLE: bool = True # 是否启用静态文件
- STATIC_URL: str = "/static" # 访问路由
- STATIC_DIR: str = "static" # 目录名
- STATIC_ROOT: Path = BASE_DIR.joinpath(STATIC_DIR) # 绝对路径
- # ================================================= #
- # ***************** 动态文件配置 ***************** #
- # ================================================= #
- UPLOAD_FILE_PATH: Path = Path("static/upload") # 上传目录
- UPLOAD_MACHINE: str = "A" # 上传机器标识
- ALLOWED_EXTENSIONS: list[str] = [ # 允许的文件类型
- ".gif",
- ".jpg",
- ".jpeg",
- ".png",
- ".ico",
- ".svg",
- ".xls",
- ".xlsx",
- ]
- MAX_FILE_SIZE: int = 10 * 1024 * 1024 # 最大文件大小(10MB)
- # ================================================= #
- # ***************** Swagger配置 ***************** #
- # ================================================= #
- SWAGGER_CSS_URL: str = "static/swagger/swagger-ui/swagger-ui.css"
- SWAGGER_JS_URL: str = "static/swagger/swagger-ui/swagger-ui-bundle.js"
- REDOC_JS_URL: str = "static/swagger/redoc/bundles/redoc.standalone.js"
- CUSTOM_CSS_URL: str = "static/swagger/custom-ui/styles.css"
- CUSTOM_JS_URL: str = "static/swagger/custom-ui/scripts.js"
- FAVICON_URL: str = "static/swagger/favicon.png"
- # ================================================= #
- # ******************* AI大模型配置 ****************** #
- # ================================================= #
- OPENAI_BASE_URL: str = ""
- OPENAI_API_KEY: str = ""
- OPENAI_MODEL: str = ""
- # ================================================= #
- # ******************* ChromaDB配置 ****************** #
- # ================================================= #
- CHROMA_PERSIST_DIR: str = str(BASE_DIR / "data" / "chroma") # ChromaDB 持久化目录
- CHROMA_COLLECTION_NAME: str = "knowledge_base" # ChromaDB 集合名称
- # ================================================= #
- # ******************* 请求限制配置 ****************** #
- # ================================================= #
- REQUEST_LIMITER_REDIS_PREFIX: str = "fastapiadmin:request_limiter:"
- # ================================================= #
- # ******************* 重构配置 ******************* #
- # ================================================= #
- @property
- def MIDDLEWARE_LIST(self) -> list[str | None]:
- """
- 根据开关组装的中间件类路径列表(未启用的项为 None)。
- 返回:
- - list[str | None]: 中间件 import 路径或 None。
- """
- # 中间件列表(注册时逆序叠加:下列第一项在列表中最前,最终位于最外层,优先生效)
- MIDDLEWARES: list[str | None] = [
- "app.core.middlewares.CustomCORSMiddleware" if self.CORS_ORIGIN_ENABLE else None,
- "app.core.middlewares.RequestLogMiddleware" if self.OPERATION_LOG_RECORD else None,
- "app.core.middlewares.CustomGZipMiddleware" if self.GZIP_ENABLE else None,
- ]
- return MIDDLEWARES
- @property
- def EVENT_LIST(self) -> list[str | None]:
- """
- 应用启动时加载的全局异步事件模块路径列表。
- 返回:
- - list[str | None]: 事件模块路径或 None。
- """
- EVENTS: list[str | None] = [
- "app.core.database.redis_connect" if self.REDIS_ENABLE else None,
- ]
- return EVENTS
- @property
- def ASYNC_DB_URI(self) -> str:
- """
- 异步 SQLAlchemy 数据库 URL。
- 返回:
- - str: 异步驱动连接串。
- 异常:
- - ValueError: 数据库类型不支持时抛出。
- """
- if self.DATABASE_TYPE not in ("mysql", "postgres", "sqlite"):
- raise ValueError(
- f"数据库驱动不支持: {self.DATABASE_TYPE}, 异步数据库请选择 mysql、postgres、sqlite"
- )
- db_connect: str = ""
- if self.DATABASE_TYPE == "mysql":
- db_connect = f"mysql+asyncmy://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}?charset=utf8mb4"
- elif self.DATABASE_TYPE == "postgres":
- db_connect = f"postgresql+asyncpg://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}"
- else:
- db_connect = f"sqlite+aiosqlite:///{self.DATABASE_NAME}.db"
- return db_connect
- @property
- def DB_URI(self) -> str:
- """
- 同步 SQLAlchemy 数据库 URL。
- 返回:
- - str: 同步驱动连接串。
- 异常:
- - ValueError: 数据库类型不支持时抛出。
- """
- if self.DATABASE_TYPE not in ("mysql", "postgres", "sqlite"):
- raise ValueError(
- f"数据库驱动不支持: {self.DATABASE_TYPE}, 同步数据库请选择 mysql、postgres、sqlite"
- )
- db_connect: str = ""
- if self.DATABASE_TYPE == "mysql":
- db_connect = f"mysql+pymysql://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}?charset=utf8mb4"
- elif self.DATABASE_TYPE == "postgres":
- db_connect = f"postgresql+psycopg://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}"
- else:
- db_connect = f"sqlite:///{self.DATABASE_NAME}.db"
- return db_connect
- @property
- def REDIS_URI(self) -> str:
- """
- Redis 连接 URL。
- 返回:
- - str: redis:// 连接串。
- """
- return f"redis://{self.REDIS_USER}:{quote_plus(self.REDIS_PASSWORD)}@{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB_NAME}"
- @property
- def FASTAPI_CONFIG(self) -> dict[str, Any]:
- """
- 创建 FastAPI 应用实例时使用的关键字参数子集。
- 返回:
- - dict[str, Any]: debug、title、responses 等配置。
- """
- return {
- "debug": self.DEBUG,
- "title": self.TITLE,
- "version": self.VERSION,
- "description": self.DESCRIPTION,
- "summary": self.SUMMARY,
- "docs_url": None,
- "redoc_url": None,
- "root_path": self.ROOT_PATH,
- "responses": {
- 200: {"description": "成功"},
- 400: {"description": "请求参数错误"},
- 401: {"description": "未认证"},
- 403: {"description": "未授权"},
- 404: {"description": "资源不存在"},
- 422: {"description": "请求参数验证错误"},
- 500: {"description": "服务器内部错误"},
- },
- }
- @lru_cache(maxsize=1)
- def get_settings() -> Settings:
- """
- 获取全局 Settings 单例(lru_cache 缓存)。
- 返回:
- - Settings: 配置实例。
- """
- return Settings()
- settings = get_settings()
|