从显式鉴权到声明式事务 — Java → Python 企业级架构演进实录

一条完整的代码重构路径：Controller显式鉴权 → 注解抽象 → 事务传播 → 隐式连接注入

📌 原始代码（双语言痛点）

• 在JAVA→python的转码期，做一个python+pg的开荒项目，尝试用共通的设计思路来手搓一下框架熟悉py，另外其实作为开荒项目也不宜直接上框架来做过度工程化，因为开荒期的需求迭代可能比你写的速度快得多。

Python 原始代码（FastAPI 风格）

# ========== Controller 层：满眼业务逻辑 + 显式鉴权 ==========
@router.post("/{tid}")
async def create_contract(tid: str, body: ContractBody, request: Request):
    # 🔴 痛点1：每个接口都要重复写显式鉴权
    user = request.state.user
    role = user.get("role", "readonly")
    allowed = {"admin", "sales_manager"}
    if role not in allowed:
        raise HTTPException(status_code=403, detail="您没有此操作的权限")
    
    # 🔴 痛点2：业务逻辑和控制器混在一起
    try:
        # 这里没有 Service 层封装，直接在 Controller 里写业务
        # 而且还要手动传递 conn
        async with get_write_conn() as conn, conn.transaction():
            await contracts_repo.insert_contract_template(
                conn, tid, body.content, user["username"]
            )
            await _write_log(
                conn, user, "create", "contract_template", tid, {"name": body.name}
            )
        return {"code": 200}
    except Exception as e:
        # 手动处理异常
        raise HTTPException(status_code=500, detail=str(e))


# ========== Repo 层：显式接收 conn 参数 ==========
async def insert_contract_template(conn, tid, content, username):
    # 🔴 痛点4：每个 Repo 函数都要显式接收 conn
    await conn.execute(
        "INSERT INTO contract_templates (id, content, created_by) VALUES ($1, $2, $3)",
        tid, content, username
    )

async def _write_log(conn, user, action, target_type, target_id, extra):
    # 🔴 同样的 conn 传递噩梦
    await conn.execute(
        "INSERT INTO logs (user_id, action, target_type, target_id, extra) VALUES ($1, $2, $3, $4, $5)",
        user["id"], action, target_type, target_id, extra
    )

Java 原始代码（Spring Boot 风格）

// ========== Controller 层：满眼业务逻辑 + 显式鉴权 ==========
@RestController
@RequestMapping("/api/contracts")
public class ContractController {
    
    @Autowired
    private ContractService contractService;
    
    @PostMapping("/{tid}")
    public ResponseEntity<?> createContract(
            @PathVariable String tid,
            @RequestBody ContractBody body,
            HttpServletRequest request) {
        
        // 🔴 痛点1：每个接口都要重复写显式鉴权
        User user = (User) request.getAttribute("user");
        String role = user.getRole();
        if (!"admin".equals(role) && !"sales_manager".equals(role)) {
            throw new RuntimeException("您没有此操作的权限");
        }
        
        // 🔴 痛点2：业务逻辑和控制器混在一起
        try {
            contractService.createContract(tid, body, user.getUsername());
            // 还要手动记录日志
            logService.writeLog(user, "create", "contract_template", tid, body.getName());
            return ResponseEntity.ok().build();
        } catch (Exception e) {
            // 还要手动处理异常
            return ResponseEntity.status(500).body(e.getMessage());
        }
    }
}

// ========== Service 层：手动传递 Connection ==========
@Service
public class ContractService {
    
    @Autowired
    private DataSource dataSource;
    
    public void createContract(String tid, ContractBody body, String username) {
        // 🔴 痛点3：每个方法都要手动获取连接、管理事务
        try (Connection conn = dataSource.getConnection()) {
            conn.setAutoCommit(false);
            try {
                // 插入合同模板
                insertContractTemplate(conn, tid, body, username);
                // 这里如果有异常，需要手动回滚
                conn.commit();
            } catch (Exception e) {
                conn.rollback();
                throw e;
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
    
    private void insertContractTemplate(Connection conn, String tid, 
                                        ContractBody body, String username) {
        // 🔴 痛点4：Repo层必须显式接收 Connection 参数
        try (PreparedStatement ps = conn.prepareStatement(
                "INSERT INTO contract_templates (id, content, created_by) VALUES (?, ?, ?)")) {
            ps.setString(1, tid);
            ps.setString(2, body.getContent());
            ps.setString(3, username);
            ps.executeUpdate();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}

原始代码问题总结

• 其实到原版基础代码到手里问题就很清楚了，写的太臃肿了，没有分层，没有抽象，显式代码重复。那抽象的第一步，当然是想到切面和自定义注解，这两个东西应该在任何后端语言里面都是通用的，切面其实本质上也只是拦截器而已。

第一章：第一步重构 — 将显式鉴权抽象成注解

目标：消灭痛点①，把散落在 Controller 每个方法的显式鉴权代码，抽象成注解/装饰器。

1.1 Java 实现：自定义权限注解 + AOP 拦截器

// ========== Step 1: 定义权限注解 ==========
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Permission {
    boolean read() default false;
    boolean write() default false;
}

// ========== Step 2: 定义 AOP 切面（统一处理鉴权） ==========
@Aspect
@Component
public class PermissionAspect {
    
    @Around("@annotation(permission)")
    public Object checkPermission(ProceedingJoinPoint pjp, Permission permission) throws Throwable {
        // 从 SecurityContext 获取当前用户
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        User user = (User) auth.getPrincipal();
        String role = user.getRole();
        
        // 检查写权限
        if (permission.write()) {
            Set<String> allowed = Set.of("admin", "sales_manager");
            if (!allowed.contains(role)) {
                throw new RuntimeException("您没有此操作的权限");
            }
        }
        // 检查读权限
        if (permission.read()) {
            // 读权限逻辑...
        }
        
        return pjp.proceed();  // 放行
    }
}

// ========== Step 3: Controller 层极度清爽 ==========
@RestController
@RequestMapping("/api/contracts")
public class ContractController {
    
    @Autowired
    private ContractService contractService;
    
    @PostMapping("/{tid}")
    @Permission(write = true)  // ✅ 一行注解搞定鉴权！
    public ResponseEntity<?> createContract(@PathVariable String tid, 
                                            @RequestBody ContractBody body) {
        // ✅ Controller 只负责接收参数、调用 Service
        contractService.createContract(tid, body);
        return ResponseEntity.ok().build();
    }
}

1.2 Python 实现：自定义装饰器 + 全局依赖

# ========== Step 1: 定义权限装饰器 ==========
def permission(read: bool = False, write: bool = False):
    def decorator(func):
        # 将权限信息挂载到函数对象上
        func._perm_read = read
        func._perm_write = write
        return func
    return decorator

# ========== Step 2: 全局依赖（统一处理鉴权） ==========
async def permission_checker(request: Request):
    # 获取当前请求的路由函数
    route = request.scope.get("route")
    if not route:
        return
    endpoint = route.endpoint
    
    # 读取函数上挂载的权限元数据
    need_write = getattr(endpoint, "_perm_write", False)
    need_read = getattr(endpoint, "_perm_read", False)
    
    # 从请求上下文中获取当前用户
    user = request.state.user
    role = user.get("role", "readonly")
    
    if need_write:
        allowed = {"admin", "sales_manager"}
        if role not in allowed:
            raise HTTPException(status_code=403, detail="您没有此操作的权限")
    
    if need_read:
        # 读权限逻辑...
        pass

# 注册全局依赖
app = FastAPI(dependencies=[Depends(permission_checker)])

# ========== Step 3: Controller 层极度清爽 ==========
@router.post("/{tid}")
@permission(write=True)  # ✅ 一行装饰器搞定鉴权！
async def create_contract(tid: str, body: ContractBody):
    # ✅ Controller 只负责接收参数、调用 Service
    await contract_service.create_contract(tid, body)
    return {"code": 200}

1.3 第一步重构成果

第二章：思考 — 注解能否传播到 Service 层？（spread 机制探讨）

背景：第一轮重构后，Controller 层的 @Permission 只在 Controller 层生效。但有些业务场景需要 Service 层也感知到权限信息。

PS:其实这个时候想偷懒了，脑子里冒出来一个鬼点子，transactional的传播机制，我能不能一个@Permission挂在controller上，然后直接让service以及后面所有被穿透调用的def把controller的注解传播过来，页面瞬间清爽了。

2.1 问题：@Permission 为什么不会自动传播？

核心原理：注解/装饰器是静态元数据，挂在当前方法上，不会随调用栈自动传递到下层。

Controller 方法 (@Permission) 
    ↓ 调用
Service 方法 (没有 @Permission)
    ↓ 调用
Repo 方法 (没有 @Permission)

@Permission 的标签永远停留在 Controller 层，Service 层完全感知不到。

2.2 方案：设计 spread 参数实现”逻辑传播”

// ========== Java 实现：spread 参数 ==========
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Permission {
    boolean read() default false;
    boolean write() default false;
    boolean spread() default false;  // 新增：是否向下传播
}

// ========== 拦截器改造 ==========
@Component
public class PermissionInterceptor {
    
    public boolean preHandle(HttpServletRequest request, HandlerMethod handler) {
        Permission perm = handler.getMethodAnnotation(Permission.class);
        if (perm != null && perm.spread()) {
            // 🔑 关键：将权限信息存入 ThreadLocal（线程上下文）
            PermissionContext.set(perm.read(), perm.write());
        }
        return true;
    }
}

// ========== Service 层 AOP 读取上下文 ==========
@Aspect
@Component
public class ServicePermissionAspect {
    
    @Around("execution(* com.service.*.*(..))")
    public Object checkServicePermission(ProceedingJoinPoint pjp) {
        // 从 ThreadLocal 读取 Controller 传下来的权限
        PermissionContext.Perm perm = PermissionContext.get();
        if (perm != null && !perm.hasWrite()) {
            throw new RuntimeException("Service层检测到无写权限");
        }
        return pjp.proceed();
    }
}

// ========== 使用 ==========
@PostMapping("/{tid}")
@Permission(write = true, spread = true)  // spread=true 让权限穿透到 Service
public ResponseEntity<?> createContract(...) {
    contractService.createContract(...);
    return ResponseEntity.ok().build();
}

# ========== Python 实现：spread 参数 + ContextVar ==========
import contextvars

# ContextVar 替代 Java 的 ThreadLocal（异步安全）
_perm_context: contextvars.ContextVar = contextvars.ContextVar('perm', default=None)

def permission(read: bool = False, write: bool = False, spread: bool = False):
    def decorator(func):
        func._perm_read = read
        func._perm_write = write
        func._perm_spread = spread
        return func
    return decorator

# ========== 全局依赖改造 ==========
async def permission_checker(request: Request):
    endpoint = request.scope.get("route").endpoint
    if hasattr(endpoint, "_perm_spread") and endpoint._perm_spread:
        # 🔑 将权限存入 ContextVar（传播到 Service 层）
        _perm_context.set({
            "read": endpoint._perm_read,
            "write": endpoint._perm_write
        })
    # 继续执行 Controller 层鉴权...

# ========== Service 层读取上下文 ==========
class ContractService:
    def create_contract(self, tid, body):
        # 从 ContextVar 读取 Controller 传下来的权限
        perm = _perm_context.get()
        if perm and not perm.get("write", False):
            raise Exception("Service层检测到无写权限")
        # 执行业务逻辑...
        pass

# ========== 使用 ==========
@router.post("/{tid}")
@permission(write=True, spread=True)  # spread=True 让权限穿透到 Service
async def create_contract(tid: str, body: ContractBody):
    await contract_service.create_contract(tid, body)
    return {"code": 200}

2.3 spread 机制的本质

Controller 方法 (@Permission(spread=true))
    ↓ 将权限写入 ThreadLocal/ContextVar（显式传递）
Service 方法
    ↓ 从 ThreadLocal/ContextVar 读取（隐式接收）
Repo 方法
    ↓ 也能读取到（如果愿意的话）

2.4 为什么不推荐 spread 机制？

• 其实传播的思路是没毛的，只是语言环境导致在py里面不适合这么干。但是尝试一下还是挺有意思，熟悉机制，运用机制，创造机制；解决问题的思路永远是最伟大的。

2.5 最佳实践：直接把注解打在 Service 层

// ✅ 推荐做法：权限注解直接打在 Service 层
@Service
public class ContractService {
    @Permission(write = true)  // 直接打在 Service 上！
    public void createContract(String tid, ContractBody body) {
        // 业务逻辑...
    }
}

# ✅ 推荐做法：装饰器直接打在 Service 层
class ContractService:
    @permission(write=True)  # 直接打在 Service 上！
    async def create_contract(self, tid, body):
        # 业务逻辑...
        pass

核心结论：Controller 层只做认证（Authentication），Service 层做授权（Authorization）。权限注解直接打在 Service 层，既清晰又避免了 spread 的隐式依赖问题。

第三章：第二步重构 — 自定义 @transactional 注解（带传播机制）

目标：消灭痛点③（手动管理连接和事务），用声明式事务替代手写 getConnection + commit/rollback。

3.1 Java 实现：Spring 原生 @Transactional

// ========== 直接用 Spring 原生注解 ==========
@Service
public class ContractService {
    
    @Autowired
    private ContractTemplateRepository templateRepo;
    
    @Autowired
    private LogRepository logRepo;
    
    @Transactional(propagation = Propagation.REQUIRED)  // ✅ 声明式事务！
    public void createContract(String tid, ContractBody body, String username) {
        // ✅ 完全不用手动获取 Connection
        // ✅ 完全不用手动 commit/rollback
        templateRepo.insert(tid, body.getContent(), username);
        logRepo.writeLog(username, "create", "contract_template", tid, body.getName());
        // 发生异常自动回滚，正常执行自动提交
    }
}

// ========== Repository 层不再接收 Connection ==========
@Repository
public class ContractTemplateRepository {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;  // Spring 自动注入，从当前事务获取连接
    
    public void insert(String tid, String content, String username) {
        // ✅ 不再需要 Connection 参数！
        jdbcTemplate.update(
            "INSERT INTO contract_templates (id, content, created_by) VALUES (?, ?, ?)",
            tid, content, username
        );
    }
}

3.2 Python 实现：自定义 @transactional 装饰器

# ========== Step 1: 上下文存储层（ContextVar） ==========
# db_context.py
import contextvars
from asyncpg import Connection

_current_conn: contextvars.ContextVar[Connection] = contextvars.ContextVar('db_conn', default=None)

def get_current_conn() -> Connection:
    conn = _current_conn.get()
    if conn is None:
        raise RuntimeError("没有活跃的数据库连接，请确保在 @transactional 装饰器内调用")
    return conn

def set_current_conn(conn: Connection):
    _current_conn.set(conn)

def reset_current_conn():
    _current_conn.set(None)

# ========== Step 2: 核心装饰器（支持传播行为） ==========
# decorators.py
from functools import wraps
from db_context import get_current_conn, set_current_conn, reset_current_conn
from connection_pool import get_write_conn

def transactional(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        existing_conn = get_current_conn()
        
        if existing_conn is not None:
            # ✅ 情况1：外层已有事务，直接复用（REQUIRED 默认行为）
            return await func(*args, **kwargs)
        else:
            # ✅ 情况2：顶层入口，新建连接 + 管理事务
            async with get_write_conn() as conn:
                set_current_conn(conn)
                try:
                    async with conn.transaction():
                        result = await func(*args, **kwargs)
                    return result
                except Exception as e:
                    raise e
                finally:
                    reset_current_conn()
    return wrapper

# ========== Step 3: Service 层使用 ==========
class ContractService:
    
    @transactional  # ✅ 一行装饰器搞定事务！
    async def create_contract(self, tid: str, body: ContractBody, username: str):
        # ✅ 完全不用手动 get_write_conn()
        # ✅ 完全不用手动 commit/rollback
        await contracts_repo.insert_contract_template(tid, body.content, username)
        await log_repo.write_log(username, "create", "contract_template", tid, body.name)
        # 发生异常自动回滚，正常执行自动提交

# ========== Step 4: 支持 REQUIRES_NEW 传播级别 ==========
def transactional(propagation: str = "REQUIRED"):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            existing_conn = get_current_conn()
            
            if propagation == "REQUIRES_NEW" and existing_conn is not None:
                # 挂起外层连接，新建独立连接
                async with get_write_conn() as new_conn:
                    old_conn = existing_conn
                    set_current_conn(new_conn)
                    try:
                        async with new_conn.transaction():
                            result = await func(*args, **kwargs)
                        return result
                    finally:
                        set_current_conn(old_conn)  # 恢复外层连接
            else:
                # REQUIRED 逻辑（同上面 Step 2）
                # ...
        return wrapper
    return decorator

# 使用示例
class LogService:
    @transactional(propagation="REQUIRES_NEW")  # 独立事务，不受外层影响
    async def write_log(self, user, action, target_type, target_id, extra):
        await log_repo.insert(user["id"], action, target_type, target_id, extra)

3.3 第二步重构成果

第四章：第三步重构 — 消灭 Repo 层的 conn 参数（自动注入上下文）

目标：消灭痛点④，让 Repo 层不再显式接收 conn 参数，实现”纯净的业务代码”。

PS:最痛恨的手动连接池，像原始人进入了计算机时代，上一次用jdbc还是在上次。

4.1 Java 实现：Spring 的 JdbcTemplate 自动注入

// ========== Repository 层：不需要 conn 参数 ==========
@Repository
public class ContractTemplateRepository {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;  // Spring 自动从当前事务获取连接
    
    public void insert(String tid, String content, String username) {
        // ✅ 完全不需要 conn 参数！JdbcTemplate 内部从 ThreadLocal 获取
        jdbcTemplate.update(
            "INSERT INTO contract_templates (id, content, created_by) VALUES (?, ?, ?)",
            tid, content, username
        );
    }
}

@Repository
public class LogRepository {
    
    @Autowired
    private JdbcTemplate jdbcTemplate;
    
    public void writeLog(String username, String action, String targetType, 
                         String targetId, String extra) {
        // ✅ 同样不需要 conn
        jdbcTemplate.update(
            "INSERT INTO logs (user_id, action, target_type, target_id, extra) VALUES (?, ?, ?, ?, ?)",
            username, action, targetType, targetId, extra
        );
    }
}

// ========== Service 层彻底纯净 ==========
@Service
public class ContractService {
    
    @Autowired
    private ContractTemplateRepository templateRepo;
    
    @Autowired
    private LogRepository logRepo;
    
    @Transactional
    public void createContract(String tid, ContractBody body, String username) {
        // ✅ 既没有 conn，也没有 TransactionManager 样板代码
        templateRepo.insert(tid, body.getContent(), username);
        logRepo.writeLog(username, "create", "contract_template", tid, body.getName());
    }
}

Spring 底层原理：JdbcTemplate 内部调用 DataSourceUtils.getConnection(dataSource)，从当前线程的 TransactionSynchronizationManager 获取事务绑定的连接。

4.2 Python 实现：db_helper 工具函数封装

# ========== Step 1: 封装 db_helper（对标 JdbcTemplate） ==========
# db_helper.py
from db_context import get_current_conn

async def execute(sql: str, *args):
    """执行 SQL（自动从上下文获取连接）"""
    conn = get_current_conn()
    return await conn.execute(sql, *args)

async def fetch_all(sql: str, *args):
    """查询多条（自动从上下文获取连接）"""
    conn = get_current_conn()
    return await conn.fetch(sql, *args)

async def fetch_one(sql: str, *args):
    """查询单条（自动从上下文获取连接）"""
    conn = get_current_conn()
    return await conn.fetchrow(sql, *args)

# ========== Step 2: Repository 层彻底干净 ==========
# contracts_repo.py
from db_helper import execute

async def insert_contract_template(tid: str, content: str, username: str):
    # ✅ 完全不需要 conn 参数！自动从上下文获取
    await execute(
        "INSERT INTO contract_templates (id, content, created_by) VALUES ($1, $2, $3)",
        tid, content, username
    )

async def find_by_tid(tid: str):
    # ✅ 同样干净
    from db_helper import fetch_one
    return await fetch_one("SELECT * FROM contract_templates WHERE id = $1", tid)

# log_repo.py
from db_helper import execute

async def write_log(user_id: str, action: str, target_type: str, 
                    target_id: str, extra: str):
    # ✅ 完全不需要 conn
    await execute(
        "INSERT INTO logs (user_id, action, target_type, target_id, extra) VALUES ($1, $2, $3, $4, $5)",
        user_id, action, target_type, target_id, extra
    )

# ========== Step 3: Service 层终极纯净 ==========
# contract_service.py
class ContractService:
    
    @transactional  # 只负责事务边界
    async def create_contract(self, tid: str, body: ContractBody, username: str):
        # ✅ 没有 async with get_write_conn()
        # ✅ 没有 conn 参数传递
        # ✅ 没有手动 commit/rollback
        await contracts_repo.insert_contract_template(tid, body.content, username)
        await log_repo.write_log(username, "create", "contract_template", tid, body.name)

4.3 第三步重构成果

第五章：最终架构总览

5.1 完整代码调用链路

┌─────────────────────────────────────────────────────────────────────────────┐
│                              HTTP 请求进入                                  │
└─────────────────────────────────────────────────────────────────────────────┘
                                    ↓
┌─────────────────────────────────────────────────────────────────────────────┐
│  Controller 层                                                             │
│  ├── @router.post("/{tid}")                                                │
│  ├── @permission(write=True)           ← 一行注解搞定鉴权                   │
│  └── await service.create_contract()   ← 纯业务调用，无任何样板代码          │
└─────────────────────────────────────────────────────────────────────────────┘
                                    ↓
┌─────────────────────────────────────────────────────────────────────────────┐
│  Service 层                                                                │
│  ├── @transactional                    ← 一行装饰器搞定事务                 │
│  │   ├── 从连接池获取连接                                                   │
│  │   ├── 绑定到 ContextVar                                                   │
│  │   ├── BEGIN TRANSACTION                                                 │
│  │   ├── 执行业务方法                                                       │
│  │   ├── COMMIT / ROLLBACK                                                │
│  │   └── 清理 ContextVar                                                   │
│  └── await repo.insert(...)             ← 无 conn 参数传递！                │
└─────────────────────────────────────────────────────────────────────────────┘
                                    ↓
┌─────────────────────────────────────────────────────────────────────────────┐
│  Repository 层                                                             │
│  ├── await execute("INSERT...")        ← 调用 db_helper                     │
│  │   └── conn = get_current_conn()     ← 从 ContextVar 隐式获取              │
│  └── await conn.execute(sql)           ← 执行业务 SQL                      │
└─────────────────────────────────────────────────────────────────────────────┘

5.2 最终代码对比

Java 最终版

// ========== Controller ==========
@RestController
@RequestMapping("/api/contracts")
public class ContractController {
    @Autowired private ContractService service;
    
    @PostMapping("/{tid}")
    @Permission(write = true)  // 鉴权注解
    public ResponseEntity<?> create(@PathVariable String tid, @RequestBody ContractBody body) {
        service.createContract(tid, body);
        return ResponseEntity.ok().build();
    }
}

// ========== Service ==========
@Service
public class ContractService {
    @Autowired private ContractTemplateRepository templateRepo;
    @Autowired private LogRepository logRepo;
    
    @Transactional  // 事务注解
    public void createContract(String tid, ContractBody body) {
        templateRepo.insert(tid, body.getContent());
        logRepo.writeLog("create", "contract_template", tid);
    }
}

// ========== Repository ==========
@Repository
public class ContractTemplateRepository {
    @Autowired private JdbcTemplate jdbcTemplate;  // 自动注入
    
    public void insert(String tid, String content) {
        jdbcTemplate.update("INSERT INTO ...", tid, content);  // 无 conn 参数
    }
}

Python 最终版

# ========== Controller ==========
@router.post("/{tid}")
@permission(write=True)  # 鉴权装饰器
async def create_contract(tid: str, body: ContractBody):
    await contract_service.create_contract(tid, body)
    return {"code": 200}

# ========== Service ==========
class ContractService:
    @transactional  # 事务装饰器
    async def create_contract(self, tid: str, body: ContractBody):
        await contracts_repo.insert_contract_template(tid, body.content)
        await log_repo.write_log("create", "contract_template", tid)

# ========== Repository ==========
# contracts_repo.py
async def insert_contract_template(tid: str, content: str):
    await execute("INSERT INTO contract_templates (id, content) VALUES ($1, $2)", tid, content)
    # ↑ 无 conn 参数！execute 内部自动从 ContextVar 获取连接

5.3 完整文件结构

project/
├── core/
│   ├── db_context.py          # ContextVar 存储当前连接（对标 ThreadLocal）
│   ├── db_helper.py           # execute/fetch_all 封装（对标 JdbcTemplate）
│   ├── decorators.py          # @transactional、@permission
│   └── connection_pool.py     # 全局连接池（对标 HikariCP）
├── repositories/
│   ├── contracts_repo.py      # 无 conn 参数，纯 SQL
│   └── log_repo.py
├── services/
│   ├── contract_service.py    # @transactional 业务编排
│   └── log_service.py         # @transactional(propagation="REQUIRES_NEW")
├── controllers/
│   └── contract_controller.py # @permission + 纯调用
└── main.py                    # FastAPI 启动 + 全局依赖注册

第六章：Java vs Python 完整对标表

第七章：演进路径总结

┌─────────────────────────────────────────────────────────────────────────────┐
│                    从痛点 → 解决方案 演进路径                                 │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  痛点①: Controller 显式鉴权重复代码                                          │
│       ↓                                                                     │
│  第1步: @Permission / @permission 注解抽象                                   │
│       ↓                                                                     │
│  思考: 注解能否传播到 Service? → spread 机制探讨                              │
│       ↓                                                                     │
│  结论: 不推荐穿透，直接打在 Service 层更干净                                  │
│                                                                             │
│  痛点②: Service 手动管理连接和事务                                            │
│       ↓                                                                     │
│  第2步: @Transactional / @transactional 声明式事务                           │
│       → 支持 REQUIRED / REQUIRES_NEW 传播                                   │
│       → 消灭手动 getConnection / commit / rollback                          │
│                                                                             │
│  痛点③: Repo 层每个方法都要传 conn 参数                                      │
│       ↓                                                                     │
│  第3步: ContextVar + db_helper 自动注入上下文                                │
│       → 对标 JdbcTemplate                                                    │
│       → Repo 层彻底干净: 只有纯 SQL                                         │
│                                                                             │
│  ✅ 最终成果: 三层架构各司其职，零样板代码                                    │
│     Controller → @permission  + 纯调用                                      │
│     Service   → @transactional + 纯业务编排                                 │
│     Repo      → execute("SQL")   + 无参数传递                               │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

附录：关键基础设施代码

A. 连接池初始化

# connection_pool.py
import asyncpg
from contextlib import asynccontextmanager

_pool: asyncpg.Pool = None

async def init_pool(dsn: str):
    global _pool
    _pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)

@asynccontextmanager
async def get_write_conn():
    async with _pool.acquire() as conn:
        yield conn

B. 全局依赖注册（main.py）

# main.py
from fastapi import FastAPI, Depends
from core.decorators import permission_checker
from core.connection_pool import init_pool

app = FastAPI(dependencies=[Depends(permission_checker)])

@app.on_event("startup")
async def startup():
    await init_pool("postgresql://...")

文档结束 ✅

这份文档完整记录了从”显式鉴权 + 手动连接”到”注解式鉴权 + 声明式事务 + 隐式连接注入”的全过程，用 Java 和 Python 双语言对比，展示了两种语言在企业级架构设计上的殊途同归。

当然我想说的是偷懒是创造之源，架构的演进和产生和偷懒的初心息息相关，合理、有效、逻辑严密的偷懒就是最佳的价值创造。