目录

前言

一、证明Generic无法shema

二、项目中遇到了什么问题，怎么解决的

 

前言

在写项目的时候，我实现了一个泛型容器，希望能够用来兼容多个实体的shema。但是这时候出问题了，我发现通过Generic容器传入的schema参数并没有什么用？

这是为什么呢？

原因就在于：泛型只是一个hint，可以帮助提示你应该输入什么类型的参数以及可以提示ID E此处应该传入什么参数，从而提醒你是否出错的参数。但是泛型（包括typing里的所有变量）并无法帮助你完成schema的功能。所以，我当初就是没有意识到这一点，才傻傻的去实现，结果没有任何作用，但是我又找到一种方法，可以拿到Generic容器中的类。

---

一、证明Generic无法shema

我们这里看一个简单的程序，然后最后我会贴出我项目中遇到的问题，并且是怎样解决的。

from typing import TypeVar, Type
T = TypeVar('T')

class Test(Generic[T]):
    def hello(self):
      my_type = T  # this is wrong!
      print( "I am {0}".format(my_type) )

Test[int]().hello()


# should print "I am int"
# however, it prints "I am ~T"

上面代码输出了 "I am ~T"，很意外。

下面代码会证明它无法schema

from typing import Generic, TypeVar
from pydantic import BaseModel


class ModelSchema(BaseModel):
    name: str
    age: int


Model = TypeVar("Model", bound=BaseModel)


class CRUDBase(Generic[Model]):
    def print_type(a: Model):
        print(a)
        print(type(a))


class Service1(CRUDBase[ModelSchema]):
    pass


Service1.print_type(18)

如果schema管用的话，Service1.print_type()里应该传入一个ModelSchema类型的参数，否则就会报错。但是当我用这段代码去运行时，仍然可以正常运行且输出结果，如下：

但是，在ID E中输入 Service1.print_type(18) 时， ID E会提醒你，你应该传入一个ModelSchema类型的参数，这就是typing hint的作用。但是，它并无法起到schema的作用。

 

二、项目中遇到了什么问题，怎么解决的

我在用FastAPI写一个后端项目，平时呢，我都是用Pydantic的BaseModel写好了schema，然后协程如下的形式来进行schema。直接就可以schema住。

from pydantic import BaseModel
from fastapi import APIRouter
from starlette.status import HTTP_201_CREATED


class PersonSchema(BaseModel):
    name: str
    age: int


@APIRouter.post("/", status_code=HTTP_201_CREATED)
def create(*, params: PersonSchema):
    pass

但是，在些项目的时候，发现每个接口都会有一套针对数据库的CRUDAPI，也就是post、get、put、delete四种。那么我就想写一个CRUDBaseAPI的类，所有要实现这类CRUD API的需求，只需要调用一下CRUDBaseAPI就可以完成，这样就可以剩下很多的代码了。

于是，我就使用如下方式实现：

from typing import TypeVar, Generic

from fastapi import APIRouter, Depends
from pydantic import BaseModel

Model = TypeVar("Model", bound=BaseModel)
CreateSchema = TypeVar("CreateSchema", bound=BaseModel)
UpdateSchema = TypeVar("UpdateSchema", bound=BaseModel)

class CRUDServiceBase(Generic[Model,CreateSchema,UpdateSchema]):
    @classmethod
    def register(cls, router:APIRouter,crud:CRUDBase):
        @APIRouter.post("/", response_model=Model):
        def create(*, database=Depends(connection), params:CreateSchema):
            pass
        
        @APIRouter.get("/", response_model=Model):
        def read(*, database=Depends(connection)):
            pass
        
        @APIRouter.put("/", response_model=Model):
        def update(*, database=Depends(connection), params:UpdateSchema):
            pass
        
        @APIRouter.post("/", response_model=Model):
        def delete(*, database=Depends(connection)):
            pass

        return router

这样，当我写一个类集成CRUDServiceBase就可以定制一个拥有CRUDAPI的接口了。如下

class CRUDEntityAPI(CRUDServiceBase[Entity, CreateEntitySchema,UpdateEntitySchema]):
    pass

router = CRUDEntity.register(CRUDEntity)

但是，这时候问题就出现了，因为所有的schema都不生效了，在调用生成的API时，即使我随便传入什么数据都是可以的，这肯定是不行的，因为shema失效，就没有意义了。

那么怎么做呢，我只有将Generic中传入的shema实体再拿出来，传入到post、get等api的chema中才可以起作用。

那么，怎么拿到Generic中的类对象呢？可以使用下面的方式。

Generic中传入的类会记录再__orig_class__.__args__中，并且是一个元组

self.__orig_class__.__args__[0] 代表取generic中第一个类，以此类推

所以，我就有了自己的改造方案，可以正常schema住了，代码如下：

from typing import TypeVar, Generic, Type

from fastapi import APIRouter, Depends
from pydantic import BaseModel

Model = TypeVar("Model", bound=BaseModel)
CreateSchema = TypeVar("CreateSchema", bound=BaseModel)
UpdateSchema = TypeVar("UpdateSchema", bound=BaseModel)

class CRUDServiceBase(Generic[Model,CreateSchema,UpdateSchema]):
    @classmethod
    def api_model_type(cls) -> Type[Model]:
        return cls.__orig_class__.__args__[0]
    
    @classmethod
    def api_create_type(cls) -> Type[CreateSchema]:
        return cls.__orig_class__.__args__[1]
    
    @classmethod
    def api_update_type(cls) -> Type[UpdateSchema]:
        return cls.__orig_class__.__args__[2]

    @classmethod
    def register(cls, router:APIRouter,crud:CRUDBase):
        api_model = cls.api_model_type()
        api_create = cls.api_create_type()
        api_update= cls.api_update_type()
        
        @APIRouter.post("/", response_model=api_model):
        def create(*, database=Depends(connection), params:api_create):
            pass

        @APIRouter.get("/", response_model=api_model):
        def read(*, database=Depends(connection)):
            pass

        @APIRouter.put("/", response_model=api_model):
        def update(*, database=Depends(connection), params:api_update):
            pass

        @APIRouter.post("/", response_model=api_model):
        def delete(*, database=Depends(connection)):
            pass

        return router

class CRUDEntityAPI(CRUDServiceBase[Entity, CreateEntitySchema,UpdateEntitySchema]):
    pass

router = CRUDEntity.register(CRUDEntity)