Python Project Architecture

Python项目架构

Core Principles

核心原则

Type hints everywhere — Pydantic for runtime, mypy for static
uv for everything — Package management, virtualenv, Python version
Ruff only — Replace Flake8 + Black + isort with single tool
src layout — All code under
```
src/
```
directory
pyproject.toml only — No setup.py, no requirements.txt
Async all the way — Once async, stay async through call chain
No backwards compatibility — Delete, don't deprecate. Change directly
LiteLLM for LLM APIs — Use LiteLLM proxy for all LLM integrations

处处使用类型提示 — 运行时用Pydantic，静态检查用mypy
全流程使用uv — 包管理、虚拟环境、Python版本管理
仅使用Ruff — 用单一工具替代Flake8 + Black + isort
src目录结构 — 所有代码放在
```
src/
```
目录下
仅用pyproject.toml — 不使用setup.py，不使用requirements.txt
全异步实现 — 一旦采用异步，整个调用链保持异步
不做向后兼容 — 直接删除，不标记废弃。直接修改
LLM API用LiteLLM — 所有LLM集成使用LiteLLM代理

No Backwards Compatibility

不做向后兼容

Delete unused code. Change directly. No compatibility layers.

python

undefined

删除未使用的代码。直接修改。不保留兼容层。

python

undefined

❌ BAD: Deprecated decorator kept around

❌ 错误示例：保留废弃装饰器

import warnings

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

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"

new_name = old_name # "为了向后兼容"

❌ BAD: Unused parameters with underscore

❌ 错误示例：保留带下划线的未使用参数

def process(_legacy_param, data): ...

❌ BAD: Version checking for old behavior

❌ 错误示例：为旧行为做版本检查

if version < "2.0": # old behavior ...

if version < "2.0": # 旧行为 ...

✅ GOOD: Just delete and update all usages

✅ 正确示例：直接删除并更新所有引用

def new_function(): ...

Then: Find & replace all old_function → new_function

然后：全局查找替换所有old_function → new_function

✅ GOOD: Remove unused parameters entirely

✅ 正确示例：完全移除未使用的参数

def process(data): ...

---

def process(data): ...

---

LiteLLM for LLM APIs

LLM API用LiteLLM

Use LiteLLM proxy. Don't call provider APIs directly.

python

undefined

使用LiteLLM代理。不要直接调用服务商API。

python

undefined

src/myapp/llm.py

from openai import AsyncOpenAI

from myapp.config import settings

from openai import AsyncOpenAI

from myapp.config import settings

Connect to LiteLLM proxy using OpenAI SDK

使用OpenAI SDK连接到LiteLLM代理

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 ""

---

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: """通过LiteLLM代理调用任意LLM。""" response = await client.chat.completions.create( model=model, # "gpt-4o", "claude-3-opus", "gemini-pro", 等 messages=[{"role": "user", "content": prompt}], ) return response.choices[0].message.content or ""

---

Quick Start

快速开始

1. Initialize Project

1. 初始化项目

bash

undefined

bash

undefined

Install uv (if not installed)

安装uv（如果未安装）

curl -LsSf https://astral.sh/uv/install.sh | sh

Create new project

创建新项目

uv init myapp cd myapp

Set Python version

设置Python版本

echo "3.12" > .python-version

Add dependencies

添加依赖

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

undefined

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

undefined

2. Apply Tech Stack

2. 应用技术栈

Layer	Recommendation
Package Manager	uv
Linting + Format	Ruff
Type Checking	mypy
Validation	Pydantic v2
Web Framework	FastAPI
Database	SQLAlchemy 2.0 + asyncpg
HTTP Client	httpx
Testing	pytest + pytest-asyncio
Logging	structlog

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

Version Strategy

版本策略

Always use latest. Never pin in templates.

toml

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

```
uv add
```
fetches latest compatible versions
```
uv.lock
```
ensures reproducible builds
```
uv sync
```
installs exact locked versions

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

toml

[project]
dependencies = [
    "fastapi",      # uv会解析到最新兼容版本
    "pydantic",
    "sqlalchemy",
]

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

3. Use Standard Structure (src layout)

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

myapp/
├── pyproject.toml         # 单一配置文件
├── uv.lock                # 锁定文件（提交到版本库）
├── .python-version        # uv使用的Python版本
├── src/
│   └── myapp/
│       ├── __init__.py
│       ├── __main__.py    # 入口文件
│       ├── main.py        # FastAPI应用
│       ├── config.py      # Pydantic配置
│       ├── models/        # Pydantic模型
│       │   ├── __init__.py
│       │   └── user.py
│       ├── services/      # 业务逻辑
│       │   ├── __init__.py
│       │   └── user.py
│       ├── repositories/  # 数据访问
│       │   ├── __init__.py
│       │   └── user.py
│       ├── api/           # HTTP层
│       │   ├── __init__.py
│       │   ├── deps.py    # 依赖项
│       │   └── routes/
│       │       ├── __init__.py
│       │       └── user.py
│       └── core/          # 共享工具
│           ├── __init__.py
│           ├── exceptions.py
│           └── logging.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py        # 测试夹具
│   └── test_user.py
└── Makefile

Architecture Layers

架构层级

main.py — FastAPI Application

main.py — FastAPI应用

python

undefined

python

undefined

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"}

undefined

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): # 启动时执行 setup_logging() yield # 关闭时执行 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"}

undefined

config.py — Pydantic Settings

config.py — Pydantic配置

python

undefined

python

undefined

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()

undefined

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_url: str = "postgresql+asyncpg://localhost/myapp"

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

settings = Settings()

undefined

models/ — Pydantic Models

models/ — Pydantic模型

python

undefined

python

undefined

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}

undefined

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}

undefined

services/ — Business Logic

services/ — 业务逻辑

python

undefined

python

undefined

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)

undefined

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)

undefined

api/routes/ — HTTP Handlers

api/routes/ — HTTP处理器

python

undefined

python

undefined

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)

undefined

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)

undefined

core/exceptions.py — Custom Exceptions

core/exceptions.py — 自定义异常

python

undefined

python

undefined

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")

from fastapi import HTTPException, status

class AppError(Exception): """应用基础异常类。"""

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}不存在: {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

FastAPI异常处理器

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}, )

---

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

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"]

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错误
    "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

测试

python

undefined

python

undefined

tests/conftest.py

import pytest from httpx import ASGITransport, AsyncClient

from myapp.main import app

@pytest.fixture async def client(): async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: yield client

import pytest from httpx import ASGITransport, AsyncClient

from myapp.main import app

@pytest.fixture async def client(): async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: yield client

tests/test_user.py

import pytest

@pytest.mark.asyncio async def test_create_user(client): response = await client.post( "/api/v1/users", json={"email": "test@example.com", "name": "Test User"}, ) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com"

@pytest.mark.asyncio async def test_get_user_not_found(client): response = await client.get("/api/v1/users/00000000-0000-0000-0000-000000000000") assert response.status_code == 404

---

import pytest

@pytest.mark.asyncio async def test_create_user(client): response = await client.post( "/api/v1/users", json={"email": "test@example.com", "name": "Test User"}, ) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com"

@pytest.mark.asyncio async def test_get_user_not_found(client): response = await client.get("/api/v1/users/00000000-0000-0000-0000-000000000000") assert response.status_code == 404

---

Makefile

makefile

.PHONY: dev test lint fmt check clean

makefile

.PHONY: dev test lint fmt check clean

Run development server

运行开发服务器

dev: uv run uvicorn myapp.main:app --reload

Run tests

运行测试

test: uv run pytest

Run tests with coverage

运行测试并生成覆盖率报告

test-cov: uv run pytest --cov=myapp --cov-report=html

Lint code

代码检查

lint: uv run ruff check src tests

Format code

代码格式化

fmt: uv run ruff format src tests uv run ruff check --fix src tests

Type check

类型检查

typecheck: uv run mypy src

Run all checks

运行所有检查

check: fmt lint typecheck test @echo "All checks passed!"

check: fmt lint typecheck test @echo "所有检查通过！"

Clean

清理

clean: rm -rf .pytest_cache .mypy_cache .ruff_cache htmlcov .coverage find . -type d -name pycache -exec rm -rf {} +

Sync dependencies

同步依赖

sync: uv sync

Upgrade dependencies

升级依赖

upgrade: uv lock --upgrade uv sync

---

upgrade: uv lock --upgrade uv sync

---

Checklist

检查清单

markdown

undefined

markdown

undefined

Project Setup

项目设置

Architecture

架构

Quality

质量保障

pytest with pytest-asyncio
Type hints everywhere
Structured logging
Error handling middleware

用pytest + pytest-asyncio做测试
处处使用类型提示
结构化日志
错误处理中间件

CI

CI集成

ruff check
ruff format --check
mypy
pytest

---

ruff检查
ruff格式检查
mypy类型检查
pytest测试

---

python-project

Original

Translation

Python Project Architecture

Python项目架构

Core Principles

核心原则

No Backwards Compatibility

不做向后兼容

❌ BAD: Deprecated decorator kept around

❌ 错误示例：保留废弃装饰器

❌ BAD: Alias for renamed functions

❌ 错误示例：为重命名的函数保留别名

❌ BAD: Unused parameters with underscore

❌ 错误示例：保留带下划线的未使用参数

❌ BAD: Version checking for old behavior

❌ 错误示例：为旧行为做版本检查

✅ GOOD: Just delete and update all usages

✅ 正确示例：直接删除并更新所有引用

Then: Find & replace all old_function → new_function

然后：全局查找替换所有old_function → new_function

✅ GOOD: Remove unused parameters entirely

✅ 正确示例：完全移除未使用的参数

LiteLLM for LLM APIs

LLM API用LiteLLM

src/myapp/llm.py

src/myapp/llm.py

Connect to LiteLLM proxy using OpenAI SDK

使用OpenAI SDK连接到LiteLLM代理

Quick Start

快速开始

1. Initialize Project

1. 初始化项目

Install uv (if not installed)

安装uv（如果未安装）

Create new project

创建新项目

Set Python version

设置Python版本

Add dependencies

添加依赖

2. Apply Tech Stack

2. 应用技术栈

Version Strategy

版本策略

3. Use Standard Structure (src layout)

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

Architecture Layers

架构层级

main.py — FastAPI Application

main.py — FastAPI应用

src/myapp/main.py

src/myapp/main.py

config.py — Pydantic Settings

config.py — Pydantic配置

src/myapp/config.py

src/myapp/config.py

models/ — Pydantic Models

models/ — Pydantic模型

src/myapp/models/user.py

src/myapp/models/user.py

services/ — Business Logic

services/ — 业务逻辑

src/myapp/services/user.py

src/myapp/services/user.py

api/routes/ — HTTP Handlers

api/routes/ — HTTP处理器

src/myapp/api/routes/user.py

src/myapp/api/routes/user.py

core/exceptions.py — Custom Exceptions

core/exceptions.py — 自定义异常

src/myapp/core/exceptions.py

src/myapp/core/exceptions.py

FastAPI exception handler

FastAPI异常处理器

pyproject.toml

pyproject.toml

Testing

测试

tests/conftest.py