microcosm-fastapi

FastAPI和microcosm之间的桥梁。提供在Python中托管微服务的最新速度，以及依赖注入。

顶级功能

• 异步兼容协议以添加新的微服务功能
• 使用业务逻辑类型提示指定API请求和响应，同时遵循强制的CRUD操作以访问数据库
• 使用最新的SQLAlchemy 1.4（仍处于测试阶段）支持异步PostgreSQL，以支持更多并发客户端用户并减少CPU阻塞请求
• 自动生成交互式文档，在开发工作期间可在localhost:5000/docs上访问
• 无遥测：本地托管文档和其他fastapi依赖项

从microcosm-flask迁移

如果您已经在使用microcosm，那么您有很大可能性正在使用microcosm-flask作为您的通信层。这个库通过尽可能使用相同的抽象和函数名来尝试提供一个相对直接的从microcosm-flask迁移的路径。然而，为了真正利用FastAPI中做出的最佳设计决策，我们还需要对我们的某些逻辑进行重构以适应新的FastAPI范式。

资源

当在flask中定义资源时，您可能习惯于定义一个用于将客户端用户所需序列化的所有内容的模式。这包括用于创建新对象的New模式、用于从数据库检索对象的Standard模式和用于从数据库层中提取可以搜索的URL参数的Search模式。这些是通过marshmallow定义的，它提供从json->python和python->json的序列化层。例如，如下所示

from microcosm_flask.paging import PageSchema
from marshmallow import Schema, fields

class NewPizzaSchema(Schema):
    toppings = fields.String(required=True)

class PizzaSchema(NewPizzaSchema):
    id: fields.UUID(required=True)

class PizzaSearchSchema(PageSchema):
    toppings = fields.String(required=False)

与棉花糖不同，FastAPI广泛使用了pydantic来提供验证层。与marshmallow相比，Pydantic是一个更现代的库。它使用Python类型提示来定义预期的字段类型，并且与marshmallow相比具有更多内置功能。将上述定义转换为Pydantic兼容的定义非常简单。请注意，我们完全移除了PizzaSearchSchema，因为此定义将指定在另一个文件中。

from microcosm_fastapi.conventions.schemas import BaseSchema
from uuid import UUID

class NewPizza(BaseSchema):
    toppings: str

class Pizza(NewPizza)
    id: UUID

与标准Python函数中的类型提示一样，除非指定了与其类型一起的Optional标志，否则参数是必需的。这将强制客户端调用者在创建新的Pizza时提供toppings。

路由

在microcosm-flask中，您的路由通常分为两个文件。您将拥有pizza/crud.py和pizza/controller.py。crud文件指定了给定命名空间支持的运算和资源。控制器将实现任何相关的业务逻辑，以在传递给后端存储之前转换输入客户端请求。例如：

@binding("pizza_v1_routes")
def configure_pizza_routes(graph):
    controller = graph.credential_pack_controller

    mappings = {
        Operation.Create: EndpointDefinition(
            func=transactional(controller.create),
            request_schema=NewPizzaSchema(),
            response_schema=PizzaSchema(),
        ),
        Operation.Retrieve: EndpointDefinition(
            func=controller.retrieve, 
            response_schema=PizzaSchema(),
        ),
        Operation.Search: EndpointDefinition(
            func=controller.search,
            request_schema=SearchPizzaSchema(),
            response_schema=PizzaSchema(),
        ),
    }
    configure_crud(graph, controller.ns, mappings)

    return controller.ns

@binding("pizza_controller")
class PizzaController(CRUDStoreAdapter):
    def __init__(self, graph):
        super().__init__(graph, graph.pizza_store)
        self.ns = Namespace(subject=Pizza, version="v1")

这种方法的缺点是，大量逻辑被抽象到CRUDStoreAdapter和configure_crud代码中。新团队成员不会立即了解创建API函数时API函数的实际外观。

我们新的路由约定中的目标是有一个提供完整真实来源的文件。此路由将包含给定数据库对象可用的所有API的显式定义。函数和响应签名的类型提示由microcosm-fastapi为您解析 - 请求将根据函数类型进行验证，响应将被序列化以适应返回类型注解。

from microcosm_fastapi.conventions.crud import configure_crud
from microcosm_fastapi.conventions.crud_adapter import CRUDStoreAdapter
from microcosm_fastapi.conventions.schemas import SearchSchema

@binding("pizza_route")
class PizzaController(CRUDStoreAdapter):
    def __init__(self, graph):
        super().__init__(graph, graph.pizza_store)

        ns = Namespace(
            subject=Pizza,
            version="v1",
        )

        mappings = {
            Operation.Create: self.create,
            Operation.Retrieve: self.retrieve,
            Operation.Search: self.search,
        }
        configure_crud(graph, ns, mappings)

    async def create(self, pizza: NewPizzaSchema) -> PizzaSchema:
        return await super()._create(pizza)

    async def retrieve(self, pizza_id: UUID) -> PizzaSchema:
        return await super()._retrieve(pizza_id)

    async def search(self, limit: int = 20, offset: int = 0) -> SearchSchema(PizzaSchema):
        return await super()._search(limit=limit, offset=offset)

按照惯例，边缘操作（例如检索/修补等）将由microcosm-fastapi自动传递感兴趣对象的UUID。此关键字参数的格式应为{snake_case(namespace object)}_id。请参见此处的示例retrieve。客户端仍然需要相应地类型提示此内容为UUID。

存储

我们在microcosm-fastapi中捆绑了一个异步兼容的PostgreSQL客户端。为了看到最大性能提升，您还需要将您的存储实例升级为异步兼容。

任何自定义实现的函数在调用超类时必须是await。

from microcosm_fastapi.database.store import StoreAsync

@binding("pizza_store")
class PizzaStore(StoreAsync):
    def __init__(self, graph):
        super().__init__(graph, Pizza)

    async def create(self, pizza):
        pizza.delivery_date = datetime.now()
        return await super().create(pizza)

在您的图中包含以下依赖项

app.use(
    "postgres",
    "session_maker_async",
    "postgres_async",
)

其他应用程序更改

创建两个新文件wsgi和wsgi_debug以分别托管生产环境和开发图

from annotation_jobs.app import create_app
graph = create_app()
app = graph.app

from annotation_jobs.app import create_app
graph = create_app(debug=True)
app = graph.app

更新您的main.py以托管

from microcosm_fastapi.runserver import main as runserver_main

def runserver():
    # This graph is just used for config parameters
    graph = create_app(debug=True, model_only=True)

    runserver_main("{application_bundle}.wsgi_debug:app", graph)

其他查找

QueryStringList -> microcosm_fastapi.conventions.parsers.SeparatedList

测试项目

我们已设置一个测试项目以演示新API在服务中部署时的外观。要开始，请创建一个新的数据库

createuser test_project
createdb -O test_project test_project_db
createdb -O test_project test_project_test_db

运行测试

pip install pytest pytest-cov pytest-asyncio

cd test-project
pytest test_project

版本升级

当您准备好合并PR时，您需要提高包的版本。您需要更新两个文件以包含您要提高的新版本

setup.py
.bumpversion.cfg

一旦您提高了版本并推送了更改，然后合并您的PR。

一旦PR已合并，从master检出最新版本，然后标记和推送

git checkout master
git pull
git tag X.X.X 
git push --tags