依赖注入

约 748 字大约 2 分钟

2026-08-31

理解IoC、DI、IoC容器

IoC: Inversion of Control，控制反转

DI：Dependency Injection，依赖注入

def check_permission(user: str):
    print(f"验证用户{user}权限")
    return True


def ali_oss_upload(user: str, file: str):
    if not check_permission(user):
        raise Exception("权限不足")
    print(f"阿里云，{file} 上传中...")


def upload_file(user: str, file: str):
    print("上传准备")
    ali_oss_upload(user, file)
    print("上传成功")


# 正常情况
upload_file("张三", "1.txt")

# 特殊情况：希望切换为腾讯云怎么办？？？
# 测试环境：希望使用假的oss上传怎么办？？？

from typing import Callable


def check_permission(user: str) -> bool:
    print(f"验证用户{user}权限")
    return True


def ali_oss_upload(user: str, file: str, has_permission: Callable):
    if not has_permission(user):
        raise Exception("权限不足")
    print(f"阿里云，{file} 上传中...")


def upload_file(user: str, file: str, oss_upload: Callable, has_permission: Callable):
    print("上传准备")
    oss_upload(user, file, has_permission)  # 注意这里传入的是函数本身
    print("上传成功")


# 正常情况
upload_file("张三", "1.txt", ali_oss_upload, check_permission)


# 特殊情况：希望切换为腾讯云怎么办？？？
def tencent_oss_upload(user: str, file: str, has_permission: Callable):
    if not has_permission(user):
        raise Exception("权限不足")
    print(f"腾讯云，{file} 上传中...")


upload_file("张三", "1.txt", tencent_oss_upload, check_permission)


# 测试环境：希望使用假的oss上传怎么办？？？
def fake_upload(user: str, file: str, has_permission: Callable):
    if not has_permission(user):
        raise Exception("权限不足")
    print(f"模拟，{file} 上传中...")


def fake_check_permission(user: str) -> bool:
    print(f"模拟：验证用户{user}权限")
    return True


upload_file("张三", "1.txt", fake_upload, fake_check_permission)

# Depends.py
from __future__ import annotations

from collections.abc import Callable
from typing import Any


class Depends:
    _overrides: dict[Callable[..., Any], Callable[..., Any]] = {}

    def __init__(self, func: Callable[..., Any]) -> None:
        self.func = func

    def __call__(self, *args: Any, **kwds: Any) -> Any:
        actual = self._overrides.get(self.func, self.func)
        return actual(*args, **kwds)

    @classmethod
    def override(
        cls,
        original: Callable[..., Any],
        replacement: Callable[..., Any],
    ) -> None:
        cls._overrides[original] = replacement

    @classmethod
    def clear_override(cls, func: Callable[..., Any] | None = None) -> None:
        if func is not None:
            cls._overrides.pop(func, None)
        else:
            cls._overrides.clear()



 # main.py
from Depends import Depends


def check_permission(user: str) -> bool:
    print(f"验证用户{user}权限")
    return True


def ali_oss_upload(user: str, file: str, has_permission=Depends(check_permission)):
    if not has_permission(user):
        raise Exception("权限不足")
    print(f"阿里云，{file} 上传中...")


def upload_file(user: str, file: str, oss_upload=Depends(ali_oss_upload)):
    print("上传准备")
    oss_upload(user, file)  # 注意这里传入的是函数本身
    print("上传成功")


# 正常情况
upload_file("张三", "1.txt")


# 特殊情况：希望切换为腾讯云怎么办？？？
def tencent_oss_upload(user: str, file: str, has_permission=Depends(check_permission)):
    if not has_permission(user):
        raise Exception("权限不足")
    print(f"腾讯云，{file} 上传中...")


Depends.override(ali_oss_upload, tencent_oss_upload)
upload_file("张三", "2.txt")


# 测试环境：希望使用假的oss上传怎么办？？？
def fake_upload(user: str, file: str, has_permission=Depends(check_permission)):
    if not has_permission(user):
        raise Exception("权限不足")
    print(f"模拟，{file} 上传中...")


def fake_check_permission(user: str) -> bool:
    print(f"模拟：验证用户{user}权限")
    return True


Depends.clear_override()
Depends.override(ali_oss_upload, fake_upload)
Depends.override(check_permission, fake_check_permission)
upload_file("张三", "3.txt")

IoC：一种设计思想，旨在将某种选择权移交出去，从而避免与其耦合

DI：将IoC思想落地的一种手段

IoC容器：优化DI的工具，它负责交付依赖项

完善API层

复制代码

复制当前目录的duyi-service下的内容到apps/web-service/app中

该写get_db

async def get_db():
    async with get_session_factory().begin() as session:
        yield session