项目模板

fastapi

1、初始化项目

#!/bin/bash
# fastapi 项目初始化

# 初始化目录结构
mkdir -pv ./{core,common,service,repo,scheme,utils,api/v1,db,etc,model}

# 增加日志
cat > core/logging.py <<'EOF'
import logging
import os,sys
from datetime import datetime
from etc.config import settings
from logging.handlers import RotatingFileHandler

# 创建日志目录
LOG_DIR = os.environ.get("LOG_DIR", settings.LOG_DIR)
if not os.path.exists(LOG_DIR): os.makedirs(LOG_DIR)


# 业务日志
BIZ_LOG_FILE_NAME = f"biz{datetime.now().strftime('%Y%m%d')}.log"
# 系统日志
SYS_LOG_FILE_NAME = f"sys{datetime.now().strftime('%Y%m%d')}.log"

# 文件路径
BIZ_LOG_FILE = os.path.join(LOG_DIR, BIZ_LOG_FILE_NAME)
SYS_LOG_FILE = os.path.join(LOG_DIR, SYS_LOG_FILE_NAME)

# 日志格式
BUSINESS_LOG_FORMAT = "[%(asctime)s] [%(name)s] [%(levelname)s] [%(filename)s:%(funcName)s:%(lineno)d] %(message)s"
SYSTEM_LOG_FORMAT = "[%(asctime)s] [%(name)s] [%(levelname)s] [%(filename)s:%(funcName)s:%(lineno)d] %(message)s"

# 业务日志
def setup_business_logger():
    # 禁用 Uvicorn 访问日志
    logging.getLogger("uvicorn.access").setLevel(logging.CRITICAL)
    logger = logging.getLogger("BIZ")
    logger.setLevel(logging.INFO)  # 业务日志一般从 INFO 开始

    # 文件 Handler（带轮转：最多10MB，保留5个旧文件）
    file_handler = RotatingFileHandler(
        BIZ_LOG_FILE, maxBytes=200 * 1024 * 1024, backupCount=5, encoding="utf-8"
    )
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(logging.Formatter(BUSINESS_LOG_FORMAT))


    # 日志输出到控制台
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(logging.Formatter(BUSINESS_LOG_FORMAT))

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

# 系统日志
def setup_system_logger():
    logger = logging.getLogger("SYS")
    logger.setLevel(logging.DEBUG)  # 系统日志可能需要更多调试信息

    # 日志输出到控制台
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(logging.DEBUG)
    # 文件 Handler（带轮转：最多10MB，保留5个旧文件）
    file_handler = RotatingFileHandler(
        SYS_LOG_FILE, maxBytes=200 * 1024 * 1024, backupCount=5, encoding="utf-8"
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(logging.Formatter(SYSTEM_LOG_FORMAT))    
    console_handler.setFormatter(logging.Formatter(SYSTEM_LOG_FORMAT))

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

# 初始化日志
biz_log = setup_business_logger()
sys_log = setup_system_logger()
EOF

# 增加配置文件
cat > etc/config.py <<'EOF'
from pydantic_settings import BaseSettings
from pydantic import SecretStr, Field
from pathlib import Path


class Config(BaseSettings):
  # 服务监听端口
  SERVER_PORT: int = Field(..., env="SERVER_PORT")

  # 日志路径
  LOG_DIR: str = Field(..., env="LOG_DIR")

  class Config:
      env_file = ".env"

settings = Config()
EOF

# 增加默认环境变量
cat > .env <<'EOF'
# 服务监听端口
SERVER_PORT=9090
# 日志路径
LOG_DIR=/tmp
EOF

# 增加路由规则
cat > api/__init__.py <<'EOF'
# -*- coding:utf-8 -*-
from fastapi import APIRouter
v1_api_router = APIRouter(prefix="/api/v1")
EOF

# 增加数据库脚本
cat > db/database.py <<'EOF'
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from etc.config import settings
from core.logging import biz_log,sys_log

#DATABASE_URL = f"mysql+pymysql://{settings.DB_USERNAME}:{settings.DB_PASSWORD}@{settings.DB_HOSTS}:{settings.DB_PORT}/{settings.DB_DATABASE}?charset=utf8mb4"
DATABASE_URL="sqlite:///./test.db"
sys_log.info(f"数据库连接信息: {DATABASE_URL}")

# 创建 SQLAlchemy 引擎
engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True
    )

# 创建 session 工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 声明模型基类
Base = declarative_base()

# 统一设置字符集
class BaseModel(Base):
    __abstract__ = True
    
    # 为所有表设置统一的字符集和排序规则
    __table_args__ = {
        'mysql_engine': 'InnoDB',
        'mysql_charset': 'utf8mb4',
        'mysql_collate': 'utf8mb4_unicode_ci'
    }
EOF

cat > db/init_db.py <<'EOF'
from db.database import Base, engine # 导入共享的 Base 和 engine
from core.logging import sys_log

# ----------------------------------------------------
# 导入所有模型文件，让 SQLAlchemy 认识它们！
# ----------------------------------------------------
# import model.book.model_book
# ... 导入您的其他模型文件

def initialize_database():
    """
    负责创建数据库文件和所有表结构。
    """
    sys_log.info("正在检查并创建数据库表结构...")
    # Base.metadata 现在包含了上面所有导入的模型信息
    Base.metadata.create_all(bind=engine)
    sys_log.info("数据库表结构检查完成.")
EOF

# 生成 .gitignore
cat >.gitignore <<'EOF'
# ----------------------------------------------------
# 1. 虚拟环境和编译文件 (Python Standard)
# ----------------------------------------------------
# 虚拟环境目录
venv/
.venv/
env/
.Python

# Python 编译和缓存文件
__pycache__/
*.pyc
*.pyo
*~
*.swp

# ----------------------------------------------------
# 2. IDE 和操作系统文件
# ----------------------------------------------------
.DS_Store    # macOS
.vscode      # Visual Studio Code 缓存和本地配置
.idea/       # PyCharm/IntelliJ 配置文件
Thumbs.db    # Windows

# ----------------------------------------------------
# 3. 敏感配置和数据库
# ----------------------------------------------------
# 【关键】包含数据库密码和 API 密钥的配置文件
.env
.flaskenv

# 本地 SQLite/测试数据库文件
*.sqlite3
*.db
sql_app.db
test.db

# ----------------------------------------------------
# 4. 测试和文档生成文件
# ----------------------------------------------------
.pytest_cache/  # pytest 缓存
.coverage       # coverage 报告
htmlcov/
docs/_build/    # Sphinx 文档生成目录

# ----------------------------------------------------
# 5. 打包和部署文件
# ----------------------------------------------------
# Docker 构建缓存和忽略文件
.dockerignore
build/
dist/
EOF

# 增加启动脚本
cat > main.py <<'EOF'
# app/main.py
import uvicorn
from fastapi import FastAPI
from api import v1_api_router
from core.logging import biz_log,sys_log
from etc.config import settings
from starlette.middleware.cors import CORSMiddleware
from db.database import Base, engine

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 创建库表结构
Base.metadata.create_all(bind=engine)

# 注册路由
app.include_router(v1_api_router)

@app.get("/")
def read_root():
    biz_log.error("Root endpoint called")
    return {"message": "Welcome to FastAPI!"}

def main():
    sys_log.info(f"服务启动在端口 {settings.SERVER_PORT}")

    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=settings.SERVER_PORT,
        workers=1,
        reload=True,
        log_config=None
    )


if __name__ == "__main__":
    main()
EOF