python-project

Python 项目架构

核心原则

• 处处使用类型提示 — 运行时使用 Pydantic，静态检查使用 mypy
• 所有操作都使用 uv — 包管理、虚拟环境、Python 版本
• 仅使用 Ruff — 用一个工具替代 Flake8、Black 和 isort
• src 布局 — 所有代码都放在 src/ 目录下
• 仅使用 pyproject.toml — 不使用 setup.py 或 requirements.txt
• 全程异步 — 一旦使用异步，就在整个调用链中保持异步
• 不考虑向后兼容 — 删除，不要弃用。直接更改
• LLM API 使用 LiteLLM — 所有 LLM 集成都使用 LiteLLM 代理

---

不考虑向后兼容

删除未使用的代码。直接更改。不添加兼容层。

# ❌ BAD: Deprecated decorator kept around
import warnings

def old_function():
    warnings.warn("Use new_function instead", DeprecationWarning)
    return new_function()

# ❌ BAD: Alias for renamed functions
new_name = old_name  # "for backwards compatibility"

# ❌ BAD: Unused parameters with underscore
def process(_legacy_param, data):
    ...

# ❌ BAD: Version checking for old behavior
if version < "2.0":
    # old behavior
    ...

# ✅ GOOD: Just delete and update all usages
def new_function():
    ...
# Then: Find & replace all old_function → new_function

# ✅ GOOD: Remove unused parameters entirely
def process(data):
    ...

---

LLM API 使用 LiteLLM

使用 LiteLLM 代理。不要直接调用提供商 API。

# src/myapp/llm.py
from openai import AsyncOpenAI

from myapp.config import settings

# Connect to LiteLLM proxy using OpenAI SDK
client = AsyncOpenAI(
    base_url=settings.litellm_url,  # "http://localhost:4000"
    api_key=settings.litellm_api_key,
)


async def complete(prompt: str, model: str = "gpt-4o") -> str:
    """Call any LLM through LiteLLM proxy."""
    response = await client.chat.completions.create(
        model=model,  # "gpt-4o", "claude-3-opus", "gemini-pro", etc.
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content or ""

---

快速开始

1. 初始化项目

# Install uv (if not installed)
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create new project
uv init myapp
cd myapp

# Set Python version
echo "3.12" > .python-version

# Add dependencies
uv add fastapi uvicorn pydantic sqlalchemy httpx
uv add --dev pytest pytest-asyncio ruff mypy

2. 应用技术栈

层	推荐方案
包管理器	uv
代码检查 + 格式化	Ruff
类型检查	mypy
验证	Pydantic v2
Web 框架	FastAPI
数据库	SQLAlchemy 2.0 + asyncpg
HTTP 客户端	httpx
测试	pytest + pytest-asyncio
日志记录	structlog

版本策略

始终使用最新版本。模板中不要固定版本。

[project]
dependencies = [
    "fastapi",      # uv resolves to latest
    "pydantic",
    "sqlalchemy",
]

• uv add 会获取最新兼容版本
• uv.lock 确保构建可复现
• uv sync 会安装锁定的确切版本

3. 使用标准结构（src 布局）

myapp/
├── pyproject.toml         # Single config file
├── uv.lock                # Lock file (commit this)
├── .python-version        # Python version for uv
├── src/
│   └── myapp/
│       ├── __init__.py
│       ├── __main__.py    # Entry point
│       ├── main.py        # FastAPI app
│       ├── config.py      # Pydantic Settings
│       ├── models/        # Pydantic models
│       │   ├── __init__.py
│       │   └── user.py
│       ├── services/      # Business logic
│       │   ├── __init__.py
│       │   └── user.py
│       ├── repositories/  # Data access
│       │   ├── __init__.py
│       │   └── user.py
│       ├── api/           # HTTP layer
│       │   ├── __init__.py
│       │   ├── deps.py    # Dependencies
│       │   └── routes/
│       │       ├── __init__.py
│       │       └── user.py
│       └── core/          # Shared utilities
│           ├── __init__.py
│           ├── exceptions.py
│           └── logging.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py        # Fixtures
│   └── test_user.py
└── Makefile

---

架构层

main.py — FastAPI 应用

# src/myapp/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI

from myapp.api.routes import router
from myapp.config import settings
from myapp.core.logging import setup_logging
from myapp.db import engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    setup_logging()
    yield
    # Shutdown
    await engine.dispose()


app = FastAPI(
    title=settings.app_name,
    lifespan=lifespan,
)

app.include_router(router, prefix="/api/v1")


@app.get("/health")
async def health():
    return {"status": "ok"}

config.py — Pydantic 设置

# src/myapp/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )

    app_name: str = "myapp"
    debug: bool = False

    # Database
    database_url: str = "postgresql+asyncpg://localhost/myapp"

    # LiteLLM
    litellm_url: str = "http://localhost:4000"
    litellm_api_key: str = ""


settings = Settings()

models/ — Pydantic 模型

# src/myapp/models/user.py
from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, EmailStr, Field


class UserBase(BaseModel):
    email: EmailStr
    name: str = Field(min_length=2, max_length=100)


class UserCreate(UserBase):
    pass


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    name: str | None = Field(default=None, min_length=2, max_length=100)


class User(UserBase):
    id: UUID
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}

services/ — 业务逻辑

# src/myapp/services/user.py
from uuid import UUID

from myapp.core.exceptions import NotFoundError, ConflictError
from myapp.models.user import User, UserCreate, UserUpdate
from myapp.repositories.user import UserRepository


class UserService:
    def __init__(self, repo: UserRepository):
        self.repo = repo

    async def get(self, id: UUID) -> User:
        user = await self.repo.get(id)
        if not user:
            raise NotFoundError("user", str(id))
        return user

    async def create(self, data: UserCreate) -> User:
        existing = await self.repo.get_by_email(data.email)
        if existing:
            raise ConflictError("email already exists")
        return await self.repo.create(data)

    async def update(self, id: UUID, data: UserUpdate) -> User:
        user = await self.get(id)
        return await self.repo.update(user, data)

    async def delete(self, id: UUID) -> None:
        user = await self.get(id)
        await self.repo.delete(user)

api/routes/ — HTTP 处理程序

# src/myapp/api/routes/user.py
from uuid import UUID

from fastapi import APIRouter, Depends, status

from myapp.api.deps import get_user_service
from myapp.models.user import User, UserCreate, UserUpdate
from myapp.services.user import UserService

router = APIRouter(prefix="/users", tags=["users"])


@router.get("/{id}", response_model=User)
async def get_user(
    id: UUID,
    service: UserService = Depends(get_user_service),
):
    return await service.get(id)


@router.post("", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(
    data: UserCreate,
    service: UserService = Depends(get_user_service),
):
    return await service.create(data)


@router.patch("/{id}", response_model=User)
async def update_user(
    id: UUID,
    data: UserUpdate,
    service: UserService = Depends(get_user_service),
):
    return await service.update(id, data)


@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    id: UUID,
    service: UserService = Depends(get_user_service),
):
    await service.delete(id)

core/exceptions.py — 自定义异常

# src/myapp/core/exceptions.py
from fastapi import HTTPException, status


class AppError(Exception):
    """Base application error."""

    def __init__(self, message: str, code: str):
        self.message = message
        self.code = code
        super().__init__(message)


class NotFoundError(AppError):
    def __init__(self, resource: str, id: str):
        super().__init__(f"{resource} not found: {id}", "NOT_FOUND")


class ConflictError(AppError):
    def __init__(self, message: str):
        super().__init__(message, "CONFLICT")


class ValidationError(AppError):
    def __init__(self, message: str):
        super().__init__(message, "VALIDATION_ERROR")


# FastAPI exception handler
def app_error_to_http(error: AppError) -> HTTPException:
    status_map = {
        "NOT_FOUND": status.HTTP_404_NOT_FOUND,
        "CONFLICT": status.HTTP_409_CONFLICT,
        "VALIDATION_ERROR": status.HTTP_400_BAD_REQUEST,
    }
    return HTTPException(
        status_code=status_map.get(error.code, status.HTTP_500_INTERNAL_SERVER_ERROR),
        detail={"message": error.message, "code": error.code},
    )

---

pyproject.toml

[project]
name = "myapp"
version = "0.1.0"
description = "My application"
requires-python = ">=3.12"
dependencies = [
    "fastapi",
    "uvicorn[standard]",
    "pydantic",
    "pydantic-settings",
    "sqlalchemy[asyncio]",
    "asyncpg",
    "httpx",
    "structlog",
]

[tool.uv]
dev-dependencies = [
    "pytest",
    "pytest-asyncio",
    "pytest-cov",
    "ruff",
    "mypy",
]

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = [
    "E",   # pycodestyle errors
    "F",   # pyflakes
    "I",   # isort
    "UP",  # pyupgrade
    "B",   # flake8-bugbear
    "SIM", # flake8-simplify
]

[tool.ruff.lint.isort]
known-first-party = ["myapp"]

[tool.mypy]
strict = true
python_version = "3.12"

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

---

扩展参考

从 ## Testing 开始的详细资料已移至 reference/extended.md，以保持本技能内容简洁。如果任务需要其中移出的示例、命令目录、检查清单、平台详情或实现模板，请加载该参考文档。