第一部分企业软件开发流程重要原则不要跳过需求和设计直接写代码。在本项目中前端原型 需求分析 shared-types 已经替你完成了大部分「需求与设计」——你要做的是读懂它们而不是无视它们自己发明字段名。第二部分项目原型分析C-AUTH-01 密码登录输入手机号11 位、密码≥6 位可选记住账号输出JWT、candidateInfophone、name、avatar、resumeStatus验证码登录、钉钉登录为 P2C-JOB-01 职位搜索筛选维度关键词、城市、薪资区间、经验、学历列表字段职位名、公司名、薪资、城市、经验、学历、标签分页page、pageSize点击跳转详情C-JOB-02 职位详情展示职位描述、任职要求、福利、公司信息操作投递、收藏、立即沟通浏览量 1异步C-RESUME-01 在线简历模块基本信息、工作经历、教育经历、项目经历、附件完整度计算规则与现有前端一致API 字段当前前端约定后端需对齐C-DELIVERY-01 投递记录按状态筛选已投递、已查看、感兴趣、不合适、待面试、已发 Offer、已接受展示职位、公司、薪资、投递时间、HR 反馈操作查看详情、继续沟通C-CHAT-01 即时沟通会话列表HR 头像、姓名、最后一条消息、时间、未读数聊天窗口文本消息、时间戳、已读状态支持 Enter 发送- 企业端boss-company-ui详细功能需求B-CERT-01 企业认证申请接口GET /api/v1/company/certification/status接口POST /api/v1/company/certification/applymultipart材料营业执照、法人身份证正反面、组织机构代码证可选B-POS-01 职位列表筛选关键词、状态、城市字段ID、职位名、部门、城市、薪资、经验、学历、状态、浏览量、沟通数、简历数操作详情、编辑、沟通、查看简历、暂停/关闭、删除分页B-POS-02 发布职位必填title、department、city、salaryMin/Max/Month、experience、education、number、description、requirements选填district、address、tags、welfare、gender提交后关联当前 HR 的 companyIdB-RES-01 简历列表本质为 投递记录 的企业视角筛选关键词、期望职位、学历、经验、投递状态展示候选人姓名、期望职位、投递状态、匹配度可 Mock操作查看详情、标记状态、发起聊天B-RES-02 简历详情展示完整简历内容快捷操作感兴趣、不合适、待面试、发 OfferB-CHAT-01 沟通管理会话列表需从后端拉取含未读数聊天窗口需 WebSocket 实时通信详细功能需求M-DASH-01 数据概览指标今日新增企业、新增职位、新增求职者、待处理举报接口GET /api/admin/dashboard/metrics更新频率可接受 5 分钟缓存M-CERT-01 企业认证审核展示资质图片放大、下载展示工商核验结果MVP 可 Mock操作通过、拒绝必填拒绝原因通过后CompanyStatus.ACTIVEM-COMP-01 企业管理筛选企业名、行业、城市、状态操作查看详情、封禁/解封封禁后该企业职位不可被搜索HR 不可登录M-POS-01 职位监管查看全平台职位操作下架、封禁封禁后求职者端不可见M-CHAT-01 会话质检展示 flagged 会话列表风险等级、关键词详情完整聊天记录、敏感词高亮处罚警告、限制聊天、临时/永久封禁第三部分搭建框架boss-api/├── main.py# FastAPI应用主入口│ ├── requirements.txt# Python依赖包│ └── app/# 应用核心代码├── __init__.py │ │ ├── models/# 数据模型层 (Tortoise ORM)│ ├── __init__.py │ ├── user.py# 用户相关模型│ ├── apis/# API路由层[接受参数,返回数据]│ ├── __init__.py │ └── user_api.py# 用户相关API│ └── schemas/# 数据验证层 (Pydantic)│ ├── __init__.py │ └── user.py# 用户请求/响应模型│ │ ├── services/# 业务服务层 (逻辑)│ ├── __init__.py │ ├── user.py# 用户相关的业务代码/方法│ │ ├── core/# 核心文件│ ├── __init__.py │ ├── database.py# 数据库连接信息├── config/# 配置文件│ ├── __init__.py │ ├── settings.py# 多环境配置数据库配置在app/core/database.py,增加代码:# app/core/database.py 数据库配置文件 这个文件定义了 Tortoise-ORM 连接 MySQL 数据库所需的所有配置信息 fromapp.config.settingsimportsettings# TORTOISE_ORM 是 Tortoise-ORM 规定的配置字典变量名# 后面用 register_tortoise 或 Aerich 时都会引用这个字典TORTOISE_ORM{# 1. 连接配置 —— 定义数据库连接信息connections:{# default 是默认连接的名字必须有一个 defaultdefault:{# engine指定数据库后端引擎MySQL 使用 tortoise.backends.mysqlengine:tortoise.backends.mysql,# credentials数据库连接凭证包含主机、端口、用户名、密码等credentials:{host:127.0.0.1,# MySQL 服务器地址port:3306,# MySQL 端口默认 3306user:root,# 数据库用户名password:123456,# 数据库密码请根据实际情况修改database:fastapi_db0719,# 数据库名称minsize:1,# 连接池最小连接数maxsize:5,# 连接池最大连接数charset:utf8mb4,# 字符集支持 emojiecho:True# 是否打印 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 打印}在main.py,加载数据库配置:fromcontextlibimportasynccontextmanagerfromfastapiimportFastAPIfromtortoiseimportTortoisefromapp.core.databaseimportTORTOISE_ORM# 定义 lifespan 生命周期函数asynccontextmanagerasyncdeflifespan(app:FastAPI): 使用 lifespan 管理应用的生命周期。 在应用启动时初始化数据库连接在应用关闭时释放连接。 这是 FastAPI 推荐的最佳实践。 # 启动时执行的代码print( 正在初始化 Tortoise-ORM...)# 启动阶段初始化 Tortoise-ORM 连接awaitTortoise.init(configTORTOISE_ORM)print(数据库连接初始化完成)yield# 应用运行阶段# 关闭阶段销毁数据库连接awaitTortoise.close_connections()print(数据库连接已关闭)appFastAPI(lifespanlifespan)app.get(/)asyncdefroot():return{message:Hello World}app.get(/hello/{name})asyncdefsay_hello(name:str):return{message:fHello{name}}依赖pipinstallpydantic-settings项目根目录/ ├── .env# 各环境共享的少量公共项├── .env.dev# 开发环境├── .env.test# 测试环境├── .env.prod# 生产环境真实密钥勿提交公开仓库├── .gitignore# 忽略 .env.dev / .env.prod 等├── main.py# 使用 settings 初始化 FastAPI└── app/ ├── config/ │ └── settings.py# ★ 本节核心└── core/ └── database.py# ★ 从 settings 读库配置在app/config/settings.pyimportos from typingimportOptional from pydantic_settingsimportBaseSettings, SettingsConfigDict class BaseAppSettings(BaseSettings): 基础配置类所有环境共享 model_configSettingsConfigDict(env_file_encodingutf-8,case_sensitiveFalse,extraignore,# 让子类继承 env 配置不会被覆盖掉这是核心修复env_file.env,)# 通用配置app_title: strFastAPI 项目脚手架app_version: str1.0.0api_prefix: str/api/v1app_description: strBoss项目的接口文档,包含求职者端,企业端,管理端class DevAppSettings(BaseAppSettings):开发环境 model_configSettingsConfigDict(env_file.env.dev)# 服务server_port: int8000debug_mode: boolTrue# 数据库db_url: strmysqlpymysql://root:123456localhost:3306/fastApiProject004db_host: strlocalhostdb_port: int3306db_user: strrootdb_password: str123456db_name: strmy-fastapidb_echo: boolTrue db_minsize: int1db_maxsize: int5DEBUG: boolTrue# 安全secret_key: strdev-secret-key-123456-pydanticjwt_token_secret_key: strdhsjjdkfjdkfrjfrjgr-278783jkdsdhjdhjsds-dsdksjdkajskajieuiwueiwhdshmxzxno9iytoken_expire_minutes: int120cors_allow_origins: list[str][*]# 阿里云OSSALIYUN_OSS_ACCESS_KEY_ID: strKEYALIYUN_OSS_ACCESS_KEY_SECRET: strKEYALIYUN_OSS_ENDPOINT: stross-cn-beijing.aliyuncs.comALIYUN_OSS_BUCKET_NAME: strfastapi-project-004# 钉钉DINGTALK_APP_KEY: strbf8ad584-a6d1-454d-8291-5e0158b4722bDINGTALK_REDIRECT_URI: strhttp://127.0.0.1:8000/third_party/dingtalk/login/callbackDINGTALK_CLIENT_ID: strDINGTALK_CLIENT_SECRET: strREDIS_HOST: str127.0.0.1REDIS_PORT: int6379REDIS_DB: int8DEFAULT_AVATAR: strhttps://img10.360buyimg.com/pcpubliccms/s1440x1440_jfs/t1/240214/16/3793/62089/65acb64bF35c090ae/4cce5ee81fae5a23.jpg.avif# 高德地图AMAP_SERVRER_KEY: strd0c0c0c0c0c0c0c0c0c0c0c0c0c0c0c0# 微信支付 V3MCH_ID: str1558950191MCH_SERIAL_NO: str34345964330B66427E0D3D28826C4993C77E631FPRIVATE_KEY_PATH: strapiclient_key.pemAPI_V3_KEY: strUDuLFDcmy5Eb6o0nTNZdu6ek4DDh4K8BAPP_ID: strwx74862e0dfcf69954DOMAIN: strhttps://api.mch.weixin.qq.comNOTIFY_DOMAIN: strhttps://xxx.ngrok.ioPARTNER_KEY: strT6m9iK73b0kn9g5v426MKfHQH7X8rKwbproperty def NATIVE_ORDER_URL(self):returnf{self.DOMAIN}/v3/pay/transactions/nativeproperty def QUERY_ORDER_URL(self):returnf{self.DOMAIN}/v3/pay/transactions/id/property def CLOSE_ORDER_URL(self):returnf{self.DOMAIN}/v3/pay/transactions/out-trade-no/%s/closeproperty def REFUND_URL(self):returnf{self.DOMAIN}/v3/refund/domestic/refundsproperty def QUERY_REFUND_URL(self):returnf{self.DOMAIN}/v3/refund/domestic/refunds/%sclass TestAppSettings(BaseAppSettings):测试环境 model_configSettingsConfigDict(env_file.env.test)server_port: int8001debug_mode: boolFalse db_url: strmysqlpymysql://root:123456localhost:3306/test_dbsecret_key: strtest-secret-key-789012-pydanticcors_allow_origins: list[str][https://test.yourdomain.com]class ProdAppSettings(BaseAppSettings): 生产环境 敏感配置 无默认值必须从环境变量或 .env.prod 读取 更安全、更规范 model_configSettingsConfigDict(env_file.env.prod)server_port: int80debug_mode: boolFalse 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: 多环境配置工厂标准写法 优先级传入参数系统环境变量默认 devenvenvor os.getenv(FASTAPI_ENV,dev)ifenvnotinSUPPORTED_ENVS: raise ValueError(f环境错误支持{SUPPORTED_ENVS})env_map{dev:DevAppSettings,test:TestAppSettings,prod:ProdAppSettings,}returnenv_map[env]()# 全局唯一配置实例settingsget_app_settings().env# 公共配置各环境都会先读再被 .env.dev / .env.test / .env.prod 覆盖APP_DESCRIPTIONBoss项目的接口文档,包含求职者端,企业端,管理端.env.dev# 开发环境 —— 对应 DevAppSettingsFASTAPI_ENVdev默认时加载APP_TITLEBOSS服务端项目SERVER_PORT8000DEBUG_MODETrueDB_HOST127.0.0.1DB_PORT3306DB_USERrootDB_PASSWORD123456DB_NAMEboss-apiDB_ECHOTrueDB_MINSIZE1DB_MAXSIZE5SECRET_KEYdev-secret-keyJWT_TOKEN_SECRET_KEYdev-jwt-secret-keyTOKEN_EXPIRE_MINUTES120CORS_ALLOW_ORIGINS[*]REDIS_HOST127.0.0.1REDIS_PORT6380REDIS_DB8# 第三方密钥课堂演示可填正式项目请用各自申请的密钥且不要提交公开仓库ALIYUN_OSS_ACCESS_KEY_IDALIYUN_OSS_ACCESS_KEY_SECRETALIYUN_OSS_BUCKET_NAMEDINGTALK_CLIENT_IDDINGTALK_CLIENT_SECRETMCH_IDAPI_V3_KEYAPP_ID修改:app/core/database.py# app/core/database.py Tortoise-ORM 数据库配置全部从 settings 读取。 from app.config.settingsimportsettings TORTOISE_ORM{connections:{default:{engine:tortoise.backends.mysql,credentials:{host:settings.db_host,port:settings.db_port,user:settings.db_user,password:settings.db_password,database:settings.db_name,minsize:settings.db_minsize,maxsize:settings.db_maxsize,charset:utf8mb4,echo:settings.db_echo,},}},apps:{models:{models:[app.models,aerich.models],default_connection:default,}},use_tz:False,timezone:Asia/Shanghai,}依赖pipinstalllogurulogging.py# app/core/logging.pyfrom loguruimportloggerimportsysimportlogging from pathlibimportPath from app.config.settingsimportsettings# 移除默认处理器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,levelDEBUGifsettings.DEBUGelseINFO,colorizeTrue)# 创建日志目录log_dirPath(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].nameINFO)# 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].nameWARNING)# 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].namein[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 相关的日志 namerecord[name].lower()messagerecord[message].lower()# 捕获 tortoise 后端相关的日志特别是包含 SQL 语句的日志return(tortoise.backendsinname orsqlinmessage orselectinmessage orinsertinmessage orupdateinmessage ordeleteinmessage orcreateinmessage oralterinmessage)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 的日志转发到 loguruclass InterceptHandler(logging.Handler):拦截标准 logging 的输出转发到 loguru def emit(self, record):# 获取对应的 loguru 级别try: levellogger.level(record.levelname).name except ValueError: levelrecord.levelno# 找到调用者信息frame, depthsys._getframe(6),6whileframe and frame.f_code.co_filenamelogging.__file__: frameframe.f_back depth1logger.opt(depthdepth,exceptionrecord.exc_info).log(level, record.getMessage())# 配置 Tortoise ORM 的 loggerdef setup_tortoise_logging():配置 Tortoise ORM 的日志输出# 拦截所有 tortoise 相关的 loggerlogging_loggers[asyncmy,tortoise,tortoise.backends,tortoise.backends.mysql,tortoise.backends.asyncpg,tortoise.backends.sqlite,]forlogger_nameinlogging_loggers: logging_loggerlogging.getLogger(logger_name)logging_logger.handlers[InterceptHandler()]logging_logger.setLevel(logging.DEBUGifsettings.db_echoelselogging.INFO)logging_logger.propagateFalse# 初始化 Tortoise 日志配置ifsettings.db_echo: setup_tortoise_logging()# 导出 logger__all__[logger,setup_tortoise_logging]在main.py导入,否则配置不会生效from contextlibimportasynccontextmanager from fastapiimportFastAPI from tortoiseimportTortoise from app.config.settingsimportsettings from app.core.databaseimportTORTOISE_ORM from app.core.loggingimportlogger# 导入即完成日志初始化asynccontextmanager async def lifespan(app: FastAPI):启动时初始化 Tortoise-ORM关闭时释放连接。 logger.info(正在初始化 Tortoise-ORM...)await Tortoise.init(configTORTOISE_ORM)logger.info(数据库连接初始化完成)yield await Tortoise.close_connections()logger.info(数据库连接已关闭)setup_tortoise_logging()appFastAPI(titlesettings.app_title,versionsettings.app_version,descriptionsettings.app_description,lifespanlifespan,)app.get(/)async def root():return{message:Hello World}app.get(/hello/{name})async def say_hello(name: str):ifnot name.strip(): logger.warning(hello 接口收到空 name)else: logger.info(问候用户: {}, name)return{message:fHello {name}}