跳至主要内容

Odoo FastAPI端点

项目描述

Beta License: LGPL-3 OCA/rest-framework Translate me on Weblate Try me on Runboat

此插件为将FastAPI框架无缝集成到Odoo提供基础。

此集成允许您使用来自FastAPI的所有优点来构建基于标准Python类型提示的自定义API,以供您的Odoo服务器使用。

什么是构建API?

API是一组可以从外部世界调用的函数。API的目标是提供一种从外部世界与您的应用程序交互的方式,而无需了解其内部工作方式。在构建API时,常见的错误是公开应用程序的所有内部函数,从而在外部世界和您的内部数据模型及业务逻辑之间创建紧密耦合。这不是一个好主意,因为这会使更改内部数据模型和业务逻辑变得非常困难,而且还会破坏外部世界。

在构建API时,您定义了外部世界和您的应用程序之间的契约。此契约由您公开的函数和接受的参数定义。此契约是API。当您更改内部数据模型和业务逻辑时,您仍然可以保持相同的API契约,因此不会破坏外部世界。即使您更改了实现,只要您保持相同的API契约,外部世界仍然可以正常工作。这是API的美丽之处,这也是为什么设计良好的API如此重要的原因。

一个好的API设计是为了稳定且易于使用。它旨在提供与特定用例相关的高级功能。它通过隐藏内部数据模型和业务逻辑的复杂性,使其易于使用。在构建API时,一个常见的错误是暴露应用程序的所有内部函数,让外部世界处理内部数据模型和业务逻辑的复杂性。别忘了,从事务的角度来看,对API函数的每次调用都是一个事务。这意味着如果特定的用例需要多次调用您的API,您应该提供一个在单个事务中完成所有工作的单个函数。这就是为什么API方法被称为高级和原子函数。

目录

使用方法

什么是使用fastapi构建API?

FastAPI是一个基于Python 3.7+的现代、快速(高性能)的Web框架,用于构建API。这个插件让您能够利用fastapi框架的优点,并与Odoo一起使用。

在开始之前,我们必须定义一些术语

  • 应用:FastAPI应用是一组路由、依赖项和其他组件的集合,可用于构建Web应用程序。

  • 路由器:路由器是一组可以挂载到应用中的路由。

  • 路由:路由是HTTP方法和路径之间的映射,并定义了当用户请求该路径时应该发生什么。

  • 依赖项:依赖项是一个可调用的函数,可用于从用户请求中获取一些信息,或在调用请求处理程序之前执行某些操作。

  • 请求:请求是一个对象,它包含用户浏览器作为HTTP请求的一部分发送的所有信息。

  • 响应:响应是一个对象,它包含用户浏览器构建结果页面所需的所有信息。

  • 处理程序:处理程序是一个函数,它接受一个请求并返回一个响应。

  • 中间件:中间件是一个函数,它接受一个请求和一个处理程序,并返回一个响应。

FastAPI框架基于以下原则

  • 快速:非常高性能,与NodeJS和Go相当(多亏了Starlette和Pydantic)。[可用的最快的Python框架之一]

  • 快速编码:通过提高开发功能的速度大约200%到300%。

  • 更少错误:减少大约40%的人类(开发者)引起的错误。

  • 直观:出色的编辑器支持。无处不在的自动完成。更少的调试时间。

  • 简单:设计用于易于使用和学习。更少的阅读文档时间。

  • 简洁:最小化代码重复。从每个参数声明中获得多个功能。更少的错误。

  • 健壮:获得可用于生产的代码。带有自动交互式文档。

  • 基于标准:基于(并且完全兼容)API的开放标准:OpenAPI(之前称为Swagger)和JSON Schema。

  • 开源:FastAPI是完全开源的,采用MIT许可。

第一步是安装fastapi插件。您可以使用以下命令完成此操作

$ pip install odoo-addon-fastapi

一旦插件安装完毕,您就可以开始构建您的API。您需要做的第一件事是创建一个新的依赖于“fastapi”的插件。例如,让我们创建一个名为my_demo_api的插件。

然后,您需要通过定义一个继承自“fastapi.endpoint”的模型来声明您的应用,并将您的应用名称添加到应用字段中。例如

from odoo import fields, models

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    app: str = fields.Selection(
        selection_add=[("demo", "Demo Endpoint")], ondelete={"demo": "cascade"}
    )

‘fastapi.endpoint’模型是所有端点的基本模型。端点实例是FastAPI应用在Odoo中的挂载点。当您创建一个新端点时,您可以在‘app’字段中定义要挂载的应用,并在‘path’字段中定义要挂载的路径。

图:: static/description/endpoint_create.png

FastAPI端点

得益于 ‘fastapi.endpoint’ 模型,您可以创建任意数量的端点,并在每个端点上挂载任意数量的应用。端点也是您可以定义您的应用配置参数的地方。一个典型的例子是在端点路径访问应用时,您想要使用的身份验证方法。

现在,您可以创建您的第一个路由器。为了做到这一点,您需要定义一个名为“demo_api_router”的全球变量到您的 fastapi_endpoint 模块中。

from fastapi import APIRouter
from odoo import fields, models

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    app: str = fields.Selection(
        selection_add=[("demo", "Demo Endpoint")], ondelete={"demo": "cascade"}
    )

# create a router
demo_api_router = APIRouter()

为了让您的路由器对您的应用可用,您需要将其添加到由您的 fastapi_endpoint 模型的 _get_fastapi_routers 方法返回的路由器列表中。

from fastapi import APIRouter
from odoo import api, fields, models

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    app: str = fields.Selection(
        selection_add=[("demo", "Demo Endpoint")], ondelete={"demo": "cascade"}
    )

    def _get_fastapi_routers(self):
        if self.app == "demo":
            return [demo_api_router]
        return super()._get_fastapi_routers()

# create a router
demo_api_router = APIRouter()

现在,您可以开始向您的路由器添加路由。例如,让我们添加一个返回合作伙伴列表的路由。

import sys
if sys.version_info >= (3, 9):
    from typing import Annotated
else:
    from typing_extensions import Annotated

from fastapi import APIRouter
from pydantic import BaseModel

from odoo import api, fields, models
from odoo.api import Environment

from odoo.addons.fastapi.dependencies import odoo_env

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    app: str = fields.Selection(
        selection_add=[("demo", "Demo Endpoint")], ondelete={"demo": "cascade"}
    )

    def _get_fastapi_routers(self):
        if self.app == "demo":
            return [demo_api_router]
        return super()._get_fastapi_routers()

# create a router
demo_api_router = APIRouter()

class PartnerInfo(BaseModel):
    name: str
    email: str

@demo_api_router.get("/partners", response_model=list[PartnerInfo])
def get_partners(env: Annotated[Environment, Depends(odoo_env)]) -> list[PartnerInfo]:
    return [
        PartnerInfo(name=partner.name, email=partner.email)
        for partner in env["res.partner"].search([])
    ]

现在,您可以启动您的 Odoo 服务器,安装您的插件,并为您的应用创建一个新的端点实例。完成之后,点击 docs url 访问您应用的交互式文档。

在尝试测试您的应用之前,您需要在端点实例上定义用于运行应用的用户。您可以通过设置 ‘user_id’ 字段来完成此操作。此信息是最重要的,因为它是您应用安全的基础。在端点实例中定义的用户将用于运行应用和访问数据库。这意味着该用户将能够访问他在 Odoo 中可以访问的所有数据。为了确保您应用的安全性,您应该创建一个新用户,仅用于运行您的应用,并且没有访问数据库的权限。

<record
      id="my_demo_app_user"
      model="res.users"
      context="{'no_reset_password': True, 'no_reset_password': True}"
  >
  <field name="name">My Demo Endpoint User</field>
  <field name="login">my_demo_app_user</field>
  <field name="groups_id" eval="[(6, 0, [])]" />
</record>

同时,您应该创建一个新组,用于定义将运行您的应用的用户访问权限。此组应包含预定义的组 ‘FastAPI Endpoint Runner’。此组定义了用户需要的最小访问权限,以便

  • 访问其所属的端点实例

  • 访问其自己的用户记录

  • 访问与其用户记录链接的合作伙伴记录

<record id="my_demo_app_group" model="res.groups">
  <field name="name">My Demo Endpoint Group</field>
  <field name="users" eval="[(4, ref('my_demo_app_user'))]" />
  <field name="implied_ids" eval="[(4, ref('fastapi.group_fastapi_endpoint_runner'))]" />
</record>

现在,您可以测试您的应用。您可以通过点击您定义的路由的“尝试一下”按钮来做到这一点。请求的结果将在“响应”部分显示,并包含合作伙伴列表。

处理 Odoo 环境

‘odoo.addons.fastapi.dependencies’ 模块提供了一组函数,您可以使用这些函数将可重用的依赖项注入到您的路由中。例如,‘odoo_env’ 函数返回当前的 Odoo 环境。您可以使用它从您的路由处理程序中访问 Odoo 模型和数据库。

import sys
if sys.version_info >= (3, 9):
    from typing import Annotated
else:
    from typing_extensions import Annotated

from odoo.api import Environment
from odoo.addons.fastapi.dependencies import odoo_env

@demo_api_router.get("/partners", response_model=list[PartnerInfo])
def get_partners(env: Annotated[Environment, Depends(odoo_env)]) -> list[PartnerInfo]:
    return [
        PartnerInfo(name=partner.name, email=partner.email)
        for partner in env["res.partner"].search([])
    ]

如您所见,您可以使用 ‘Depends’ 函数将依赖项注入到您的路由处理程序中。由 ‘fastapi’ 框架提供的 ‘Depends’ 函数。您可以使用它将任何依赖项注入到您的路由处理程序中。由于您的处理程序是一个 Python 函数,获取对 Odoo 环境的访问的唯一方法是将它作为依赖项注入。FastAPI 插件提供了一组可以用作依赖项的函数

  • ‘odoo_env’:返回当前的 Odoo 环境。

  • ‘fastapi_endpoint’:返回当前的 FastAPI 端点模型实例。

  • ‘authenticated_partner’:返回已验证的合作伙伴。

  • ‘authenticated_partner_env’:将当前带有已认证合作伙伴ID的Odoo环境返回到上下文中。

默认情况下,‘odoo_env’‘fastapi_endpoint’ 依赖项无需额外工作即可使用。

依赖注入机制

‘odoo_env’ 依赖项依赖于一个简单的实现,该实现从请求处理开始时由特定请求分发程序在请求处理过程中初始化的ContextVar变量中检索当前Odoo环境。

‘fastapi_endpoint’ 依赖项依赖于由 ‘fastapi’ 模块提供的 ‘dependency_overrides’ 机制。(有关依赖注入机制的更多详细信息,请参阅fastapi文档)。如果您查看‘fastapi_endpoint’依赖项的当前实现,您会看到该方法依赖于两个参数:‘endpoint_id’‘env’。这些参数本身也是依赖项。

def fastapi_endpoint_id() -> int:
    """This method is overriden by default to make the fastapi.endpoint record
    available for your endpoint method. To get the fastapi.endpoint record
    in your method, you just need to add a dependency on the fastapi_endpoint method
    defined below
    """


def fastapi_endpoint(
    _id: Annotated[int, Depends(fastapi_endpoint_id)],
    env: Annotated[Environment, Depends(odoo_env)],
) -> "FastapiEndpoint":
    """Return the fastapi.endpoint record"""
    return env["fastapi.endpoint"].browse(_id)

如您所见,这些依赖项中的一个依赖项是 ‘fastapi_endpoint_id’ 依赖项,它没有具体的实现。该方法用作在创建fastapi应用程序时必须实现/提供的合同。这就是依赖注入机制的力量所在。

如果您查看 ‘FastapiEndpoint’ 模型的 ‘_get_app’ 方法,您会看到 ‘fastapi_endpoint_id’ 依赖项被注册的一个特定方法覆盖,该方法返回原始方法当前fastapi端点模型实例的ID。

def _get_app(self) -> FastAPI:
    app = FastAPI(**self._prepare_fastapi_endpoint_params())
    for router in self._get_fastapi_routers():
        app.include_router(prefix=self.root_path, router=router)
    app.dependency_overrides[dependencies.fastapi_endpoint_id] = partial(
        lambda a: a, self.id
    )

这种机制非常强大,允许您将任何依赖项注入到您的路由处理器中,并且还可以定义一个抽象依赖项,该依赖项可以由任何其他插件使用,其实施可以取决于端点配置。

认证机制

为了使我们的应用程序不紧密耦合于特定的认证机制,我们将使用 ‘authenticated_partner’ 依赖项。对于 ‘fastapi_endpoint’,此依赖项依赖于一个抽象依赖项。

在定义路由处理器时,您可以将 ‘authenticated_partner’ 依赖项注入为路由处理器的一个参数。

from odoo.addons.base.models.res_partner import Partner


@demo_api_router.get("/partners", response_model=list[PartnerInfo])
def get_partners(
    env: Annotated[Environment, Depends(odoo_env)], partner: Annotated[Partner, Depends(authenticated_partner)]
) -> list[PartnerInfo]:
    return [
        PartnerInfo(name=partner.name, email=partner.email)
        for partner in env["res.partner"].search([])
    ]

在这个阶段,您的处理器并未绑定到特定的认证机制,但仅期望获得一个合作伙伴作为依赖项。根据您的需要,您可以为您的应用程序实现不同的认证机制。fastapi插件提供了一个默认的认证机制,使用‘BasicAuth’方法。此认证机制在 ‘odoo.addons.fastapi.dependencies’ 模块中实现,并依赖于由 ‘fastapi.security’ 模块提供的功能。

def authenticated_partner(
    env: Annotated[Environment, Depends(odoo_env)],
    security: Annotated[HTTPBasicCredentials, Depends(HTTPBasic())],
) -> "res.partner":
    """Return the authenticated partner"""
    partner = env["res.partner"].search(
        [("email", "=", security.username)], limit=1
    )
    if not partner:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Basic"},
        )
    if not partner.check_password(security.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Basic"},
        )
    return partner

如您所见,‘authenticated_partner’ 依赖项依赖于由 ‘fastapi.security’ 模块提供的 ‘HTTPBasic’ 依赖项。在这个示例实现中,我们仅检查提供的凭据是否可以用于在Odoo中认证用户。如果认证成功,我们返回与已认证用户关联的合作伙伴记录。

在某些情况下,您可能需要实现一个更复杂的身份验证机制,该机制可以依赖于令牌或会话。在这种情况下,您可以覆盖 ‘authenticated_partner’ 依赖项,通过注册一个返回已验证合作伙伴的特定方法。此外,您还可以在 fastapi 端点模型实例上使它可配置。

要实现它,您只需要为您的每个身份验证机制实现一个特定方法,并允许用户在创建新的 fastapi 端点时选择这些方法之一。假设我们想允许使用 API 密钥或基本认证进行认证。由于基本认证已经实现,我们将只实现 API 密钥认证机制。

from fastapi.security import APIKeyHeader

def api_key_based_authenticated_partner_impl(
    api_key: Annotated[str, Depends(
        APIKeyHeader(
            name="api-key",
            description="In this demo, you can use a user's login as api key.",
        )
    )],
    env: Annotated[Environment, Depends(odoo_env)],
) -> Partner:
    """A dummy implementation that look for a user with the same login
    as the provided api key
    """
    partner = env["res.users"].search([("login", "=", api_key)], limit=1).partner_id
    if not partner:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect API Key"
        )
    return partner

至于 ‘BasicAuth’ 身份验证机制,我们也依赖于 ‘fastapi.security’ 模块提供的原生安全依赖项之一。

现在我们已经为我们的两种身份验证机制实现了实现,我们可以在 fastapi 端点模型上添加一个选择字段,使用户可以选择这些身份验证机制之一。

from odoo import fields, models

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    app: str = fields.Selection(
      selection_add=[("demo", "Demo Endpoint")], ondelete={"demo": "cascade"}
    )
    demo_auth_method = fields.Selection(
        selection=[("api_key", "Api Key"), ("http_basic", "HTTP Bacic")],
        string="Authenciation method",
    )

现在我们已经有一个允许用户选择身份验证方法的选择字段,我们可以在应用程序实例化时使用依赖项覆盖机制来提供正确的 ‘authenticated_partner’ 依赖项实现。

from odoo.addons.fastapi.dependencies import authenticated_partner
class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    app: str = fields.Selection(
      selection_add=[("demo", "Demo Endpoint")], ondelete={"demo": "cascade"}
    )
    demo_auth_method = fields.Selection(
        selection=[("api_key", "Api Key"), ("http_basic", "HTTP Bacic")],
        string="Authenciation method",
    )

  def _get_app(self) -> FastAPI:
      app = super()._get_app()
      if self.app == "demo":
          # Here we add the overrides to the authenticated_partner_impl method
          # according to the authentication method configured on the demo app
          if self.demo_auth_method == "http_basic":
              authenticated_partner_impl_override = (
                  authenticated_partner_from_basic_auth_user
              )
          else:
              authenticated_partner_impl_override = (
                  api_key_based_authenticated_partner_impl
              )
          app.dependency_overrides[
              authenticated_partner_impl
          ] = authenticated_partner_impl_override
      return app

要了解依赖项覆盖机制的工作原理,您可以查看 fastapi 插件提供的演示应用程序。如果您在 fastapi 端点表单视图中选择应用程序 'demo',您将看到身份验证方法是可配置的。您还可以看到,根据您在 fastapi 端点上配置的身份验证方法,文档将发生变化。

管理应用程序配置参数

如前一小节所示,您可以在 fastapi 端点模型上添加配置字段,以允许用户配置您的应用程序(如任何扩展的 odoo 模型)。当您需要在路由处理程序中访问这些配置字段时,您可以使用 ‘odoo.addons.fastapi.dependencies.fastapi_endpoint’ 依赖项方法检索与当前请求关联的 ‘fastapi.endpoint’ 记录。

from pydantic import BaseModel, Field
from odoo.addons.fastapi.dependencies import fastapi_endpoint

class EndpointAppInfo(BaseModel):
  id: str
  name: str
  app: str
  auth_method: str = Field(alias="demo_auth_method")
  root_path: str

  class Config:
      orm_mode = True

  @demo_api_router.get(
      "/endpoint_app_info",
      response_model=EndpointAppInfo,
      dependencies=[Depends(authenticated_partner)],
  )
  async def endpoint_app_info(
      endpoint: Annotated[FastapiEndpoint, Depends(fastapi_endpoint)],
  ) -> EndpointAppInfo:
      """Returns the current endpoint configuration"""
      # This method show you how to get access to current endpoint configuration
      # It also show you how you can specify a dependency to force the security
      # even if the method doesn't require the authenticated partner as parameter
      return EndpointAppInfo.from_orm(endpoint)

fastapi 端点的某些配置字段可能会影响应用程序的实例化方式。例如,在前一小节中,我们已经看到配置在 ‘fastapi.endpoint’ 记录上的身份验证方法用于在应用程序实例化时提供正确的 ‘authenticated_partner’ 实现。为确保在修改用于应用程序实例化的配置元素时重新实例化应用程序,您必须覆盖 ‘_fastapi_app_fields’ 方法,将影响应用程序实例化的字段名称添加到返回的列表中。

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    app: str = fields.Selection(
      selection_add=[("demo", "Demo Endpoint")], ondelete={"demo": "cascade"}
    )
    demo_auth_method = fields.Selection(
        selection=[("api_key", "Api Key"), ("http_basic", "HTTP Bacic")],
        string="Authenciation method",
    )

    @api.model
    def _fastapi_app_fields(self) -> List[str]:
        fields = super()._fastapi_app_fields()
        fields.append("demo_auth_method")
        return fields

处理语言

FastAPI插件解析请求中的Accept-Language头,以确定要使用的语言。这种解析遵循RFC 7231规范。这意味着语言是由头中第一个被Odoo支持的语言(注意优先级顺序)确定的。如果头中没有找到语言,则使用Odoo默认语言。然后使用这种语言初始化由路由处理器使用的Odoo环境上下文。所有这些都使得语言管理变得非常简单。你不必担心这些。此功能还默认记录在应用程序生成的openapi文档中,以指导API消费者如何请求特定语言。

如何扩展现有应用程序

当你开发一个FastAPI应用程序时,在原生Python应用程序中无法扩展现有应用程序。这种限制不适用于FastAPI插件,因为FastAPI端点模型被设计为可扩展的。然而,扩展现有应用程序的方法与扩展Odoo模型的方法不同。

首先,重要的是要注意,当你定义一个路由时,你实际上是在客户端和服务器之间定义一个合同。这个合同由路由路径、方法(GET、POST、PUT、DELETE等)、参数和响应定义。如果你想扩展现有应用程序,你必须确保合同没有被破坏。对合同的任何更改都将遵循Liskov替换原则。这意味着客户端不应受到影响。

这实际上意味着什么?这意味着你不能更改现有路由的路径或方法。你不能更改参数的名称或响应的类型。你不能添加新的参数或新的响应。你不能删除参数或响应。如果你想更改合同,你必须创建一个新的路由。

你可以更改什么?

  • 你可以更改路由处理器的实现。

  • 你可以覆盖路由处理器的依赖项。

  • 你可以添加一个新的路由处理器。

  • 你可以扩展作为路由处理器参数或响应使用的模型。

让我们看看如何做到这一点。

更改路由处理器的实现

假设你想更改路由处理器“/demo/echo”的实现。由于路由处理器只是一个Python方法,这似乎是一项繁琐的任务,因为我们不在模型方法中,因此无法利用Odoo继承机制。

然而,FastAPI插件提供了一种方法来实现这一点。多亏了“odoo_env”依赖项方法,你可以访问当前的Odoo环境。有了这个环境,你可以访问注册表,因此你可以访问想要委派实现的模型。如果你想更改路由处理器“/demo/echo”的实现,你唯一要做的就是从定义实现的模型继承,并覆盖“echo”方法。

from pydantic import BaseModel
from fastapi import Depends, APIRouter
from odoo import models
from odoo.addons.fastapi.dependencies import odoo_env

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    def _get_fastapi_routers(self) -> List[APIRouter]:
        routers = super()._get_fastapi_routers()
        routers.append(demo_api_router)
        return routers

demo_api_router = APIRouter()

@demo_api_router.get(
    "/echo",
    response_model=EchoResponse,
    dependencies=[Depends(odoo_env)],
)
async def echo(
    message: str,
    odoo_env: Annotated[Environment, Depends(odoo_env)],
) -> EchoResponse:
    """Echo the message"""
    return EchoResponse(message=odoo_env["demo.fastapi.endpoint"].echo(message))

class EchoResponse(BaseModel):
    message: str

class DemoEndpoint(models.AbstractModel):

    _name = "demo.fastapi.endpoint"
    _description = "Demo Endpoint"

    def echo(self, message: str) -> str:
        return message

class DemoEndpointInherit(models.AbstractModel):

    _inherit = "demo.fastapi.endpoint"

    def echo(self, message: str) -> str:
        return f"Hello {message}"

覆盖路由处理器的依赖项

正如您之前所看到的,fastapi 的依赖注入机制非常强大。通过设计您的路由处理程序以依赖于具有特定功能范围的依赖项,您可以在不修改路由处理程序的情况下轻松更改依赖项的实现。在这种设计下,您甚至可以定义必须由具体应用程序实现的抽象依赖项。这正是我们之前示例中的 ‘authenticated_partner’ 依赖项的情况。(您可以在文件 ‘odoo/addons/fastapi/dependencies.py’ 中找到此依赖项的实现,以及在文件 ‘odoo/addons/fastapi/models/fastapi_endpoint_demo.py’ 中的使用情况)

添加新的路由处理程序

假设您想要添加一个新的路由处理程序 ‘/demo/echo2’。您可能会倾向于通过导入现有应用的路由器并将新的路由处理程序添加到其中来在您的新的插件中添加这个新的路由处理程序。

from odoo.addons.fastapi.models.fastapi_endpoint_demo import demo_api_router

@demo_api_router.get(
    "/echo2",
    response_model=EchoResponse,
    dependencies=[Depends(odoo_env)],
)
async def echo2(
    message: str,
    odoo_env: Annotated[Environment, Depends(odoo_env)],
) -> EchoResponse:
    """Echo the message"""
    echo = odoo_env["demo.fastapi.endpoint"].echo2(message)
    return EchoResponse(message=f"Echo2: {echo}")

这种方法的问题在于,即使该应用被调用的是您的新的插件未安装的不同数据库,您也会无条件地将新的路由处理程序添加到现有应用中。

解决方案是在您的新的插件中定义一个新的路由器,并将其添加到从您正在继承的模型 ‘fastapi.endpoint’ 的方法 ‘_get_fastapi_routers’ 返回的路由器列表中。

class FastapiEndpoint(models.Model):

    _inherit = "fastapi.endpoint"

    def _get_fastapi_routers(self) -> List[APIRouter]:
        routers = super()._get_fastapi_routers()
        if self.app == "demo":
            routers.append(additional_demo_api_router)
        return routers

additional_demo_api_router = APIRouter()

@additional_demo_api_router.get(
    "/echo2",
    response_model=EchoResponse,
    dependencies=[Depends(odoo_env)],
)
async def echo2(
    message: str,
    odoo_env: Annotated[Environment, Depends(odoo_env)],
) -> EchoResponse:
    """Echo the message"""
    echo = odoo_env["demo.fastapi.endpoint"].echo2(message)
    return EchoResponse(message=f"Echo2: {echo}")

这样,只有当应用被调用的是安装了您的新的插件的数据库名时,新的路由器才会添加到您的应用的路由器列表中。

扩展用作路由处理程序参数或响应的模型

fastapi Python 库使用 pydantic 库来定义模型。默认情况下,一旦定义了一个模型,就不能扩展它。然而,一个名为 extendable_pydantic 的配套 Python 库提供了一个使用继承与 pydantic 模型扩展现有模型的方法。如果单独使用,您需要负责告诉这个库要应用于模型的所有扩展及其应用的顺序。这并不太方便。幸运的是,有一个专门的 Odoo 插件可以使这个过程完全透明。这个插件称为 odoo-addon-extendable-fastapi

当您想要允许其他插件扩展 pydantic 模型时,您必须首先使用专用元类将模型定义为可扩展模型

from pydantic import BaseModel
from extendable_pydantic import ExtendableModelMeta

class Partner(BaseModel, metaclass=ExtendableModelMeta):
  name = 0.1

与任何其他 pydantic 模型一样,您现在可以使用此模型作为路由处理程序的参数或响应。您还可以使用使用 pydantic 定义的模型的所有功能。

@demo_api_router.get(
    "/partner",
    response_model=Location,
    dependencies=[Depends(authenticated_partner)],
)
async def partner(
    partner: Annotated[ResPartner, Depends(authenticated_partner)],
) -> Partner:
    """Return the location"""
    return Partner.from_orm(partner)

如果您需要向模型 ‘Partner’ 中添加新字段,您可以在新的插件中通过定义一个新的模型来扩展它,该模型继承自模型 ‘Partner’

from typing import Optional
from odoo.addons.fastapi.models.fastapi_endpoint_demo import Partner

class PartnerExtended(Partner, extends=Partner):
    email: Optional[str]

如果您的新的插件安装在一个数据库中,对路由处理程序 ‘/demo/partner’ 的调用将返回包含新字段 ‘email’ 的响应,如果 Odoo 记录提供了值。

{
  "name": "John Doe",
  "email": "jhon.doe@acsone.eu"
}

如果您的新的插件未安装在一个数据库中,对路由处理程序 ‘/demo/partner’ 的调用将仅返回合作伙伴的名称。

{
  "name": "John Doe"
}

在路由处理程序中管理安全性

默认情况下,路由处理器使用在“fastapi.endpoint”模型实例上配置的用户进行处理(默认为公共用户)。您之前已经看到了如何定义一个依赖项,该依赖项将用于强制执行合作伙伴的认证。当一个方法依赖于此依赖项时,“authenticated_partner_id”键会被添加到合作伙伴环境的上下文中。(如果您不需要将合作伙伴作为依赖项,但需要获取一个包含已认证用户的环境的,可以使用“authenticated_partner_env”依赖项而不是“authenticated_partner”)。

fastapi插件扩展了“ir.rule”模型,以便在安全规则的评估上下文中添加包含已认证合作伙伴id的“authenticated_partner_id”键。

如前一部分简要介绍的那样,当您开发一个fastapi应用并希望以高效和可追溯的方式保护您的数据时,一个好的做法是

  • 创建一个针对该应用的新用户,但没有任何访问权限。

  • 创建一个针对该应用的安全组,并将用户添加到该组中。(此组必须隐含“AFastAPI Endpoint Runner”组,该组提供最小访问权限)

  • 对于您想要保护的每个模型

    • 为模型添加一个“ir.model.access”记录,以允许对您的模型进行读取访问,并将组添加到记录中。

    • 为模型创建一个新的“ir.rule”记录,该记录通过在规则的域中使用“authenticated_partner_id”键来限制对模型记录的访问,以限制到已认证的合作伙伴(或者如果方法是公开的,则限制到在“fastapi.endpoint”模型实例上定义的用户)。

  • 当您需要访问已认证合作伙伴或确保服务由已认证合作伙伴调用时,在您的处理器中添加对“authenticated_partner”的依赖项。

<record
      id="my_demo_app_user"
      model="res.users"
      context="{'no_reset_password': True, 'no_reset_password': True}"
  >
  <field name="name">My Demo Endpoint User</field>
  <field name="login">my_demo_app_user</field>
  <field name="groups_id" eval="[(6, 0, [])]" />
</record>

<record id="my_demo_app_group" model="res.groups">
  <field name="name">My Demo Endpoint Group</field>
  <field name="users" eval="[(4, ref('my_demo_app_user'))]" />
  <field name="implied_ids" eval="[(4, ref('group_fastapi_endpoint_runner'))]" />
</record>

<!-- acl for the model 'sale.order' -->
<record id="sale_order_demo_app_access" model="ir.model.access">
  <field name="name">My Demo App: access to sale.order</field>
  <field name="model_id" ref="model_sale_order"/>
  <field name="group_id" ref="my_demo_app_group"/>
  <field name="perm_read" eval="True"/>
  <field name="perm_write" eval="False"/>
  <field name="perm_create" eval="False"/>
  <field name="perm_unlink" eval="False"/>
</record>

<!-- a record rule to allows the authenticated partner to access only its sale orders -->
<record id="demo_app_sale_order_rule" model="ir.rule">
  <field name="name">Sale Order Rule</field>
  <field name="model_id" ref="model_sale_order"/>
  <field name="domain_force">[('partner_id', '=', authenticated_partner_id)]</field>
  <field name="groups" eval="[(4, ref('my_demo_app_group'))]"/>
</record>

如何测试您的fastapi应用

多亏了starlette测试客户端,可以非常简单的方式测试您的fastapi应用。使用测试客户端,您可以像调用真实的HTTP端点一样调用您的路由处理器。测试客户端在“fastapi.testclient”模块中可用。

依赖注入机制再次发挥作用,允许您将特定于依赖项的实现注入到测试客户端中,通常由fastapi应用的正常请求处理提供。(例如,您可以将“authenticated_partner”依赖项的模拟注入到测试客户端中,以测试当合作伙伴未认证时路由处理器的行为,您也可以注入odoo_env等模拟)。

fastapi插件为测试用例提供了一个基类,您可以使用它来编写您的测试。此基类是“odoo.fastapi.tests.common.FastAPITransactionCase”。此类主要提供了方法“_create_test_client”,您可以使用它来为您的fastapi应用创建测试客户端。此方法封装了测试客户端的创建和依赖项的注入。它还确保Odoo环境在路由处理器的上下文中可用。此方法设计用于在您需要测试应用或需要测试特定路由时使用(因此,在未提供fastapi端点的插件中定义路由器的测试非常容易)。

有了这个基类,编写一个路由处理器的测试就像

from odoo.fastapi.tests.common import FastAPITransactionCase

from odoo.addons.fastapi import dependencies
from odoo.addons.fastapi.routers import demo_router

class FastAPIDemoCase(FastAPITransactionCase):

    @classmethod
    def setUpClass(cls) -> None:
        super().setUpClass()
        cls.default_fastapi_running_user = cls.env.ref("fastapi.my_demo_app_user")
        cls.default_fastapi_authenticated_partner = cls.env["res.partner"].create({"name": "FastAPI Demo"})

    def test_hello_world(self) -> None:
        with self._create_test_client(router=demo_router) as test_client:
            response: Response = test_client.get("/demo/")
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertDictEqual(response.json(), {"Hello": "World"})

在先前的示例中,我们为demo_router创建了一个测试客户端。我们也可以创建整个应用的测试客户端,而不指定路由器,而是指定应用。

from odoo.fastapi.tests.common import FastAPITransactionCase

from odoo.addons.fastapi import dependencies
from odoo.addons.fastapi.routers import demo_router

class FastAPIDemoCase(FastAPITransactionCase):

    @classmethod
    def setUpClass(cls) -> None:
        super().setUpClass()
        cls.default_fastapi_running_user = cls.env.ref("fastapi.my_demo_app_user")
        cls.default_fastapi_authenticated_partner = cls.env["res.partner"].create({"name": "FastAPI Demo"})

    def test_hello_world(self) -> None:
        demo_endpoint = self.env.ref("fastapi.fastapi_endpoint_demo")
        with self._create_test_client(app=demo_endpoint._get_app()) as test_client:
            response: Response = test_client.get(f"{demo_endpoint.root_path}/demo/")
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertDictEqual(response.json(), {"Hello": "World"})

开发fastapi应用时的整体考虑

开发fastapi应用需要遵循一些良好的做法,以确保应用是健壮且易于维护的。以下是一些:

  • 路由处理器必须尽可能简单。它不能包含任何业务逻辑。业务逻辑必须实现到服务层。路由处理器必须仅调用服务层并返回服务层的结果。为了方便扩展业务逻辑,您的服务层可以实现为可以被其他插件继承的Odoo抽象模型。

  • 路由处理器不应暴露Odoo的内部数据结构和API。它应提供客户端需要的API。更广泛地说,一个应用程序提供一组服务,这些服务针对一个定义良好的功能领域的一组特定用例。您必须始终记住,即使您升级Odoo版本或修改业务逻辑,您的API也将保持不变。

  • 路由处理器是一个事务性工作单元。在设计API时,您必须确保一个用例的完整性由一个单独的事务保证。如果您需要执行多个事务以完成一个用例,您会引入数据不一致的风险或使客户端代码更加复杂。

  • 正确处理错误。当发生错误时,路由处理器必须返回适当的错误响应。错误响应必须与API的其他部分保持一致。错误响应必须在API文档中记录。默认情况下,‘odoo-addon-fastapi’模块处理在< strong>‘odoo.exceptions’模块中定义的常见异常类型,并返回具有相应HTTP状态码的正确错误响应。路由处理器中的错误必须始终返回一个与200不同的HTTP状态码的错误响应。错误响应必须包含一个可供用户显示的易读消息。错误响应还可以包含一个可供客户端用来以特定方式处理错误的机器可读代码。

  • 当您通过pydantic模型设计JSON文档时,您必须使用适当的数据类型。例如,您必须使用数据类型 ‘datetime.date’ 来表示日期而不是字符串。您还必须正确定义字段的约束。例如,如果字段是可选的,您必须使用数据类型 ‘typing.Optional’pydantic 提供了您正确定义JSON文档所需的一切。

  • 始终为您的路由处理器使用合适的pydantic模型作为请求和/或响应。pydantic模型的字段约束必须适用于特定用例。例如,如果您的路由处理器用于创建销售订单,pydantic模型不得包含字段‘id’,因为销售订单的id将由路由处理器生成。但如果之后需要id,响应的pydantic模型必须包含字段‘id’作为必需字段。

  • 在您的JSON文档中使用描述性属性名称。例如,避免使用提供键值对扁平列表的文档。

  • 在您的JSON文档中字段命名保持一致。例如,如果您使用‘id’来表示销售订单的id,您必须使用‘id’来表示其他所有对象的id。

  • 在您的字段命名风格上保持一致。始终首选下划线而不是驼峰式。

  • 对于包含项目列表的字段名称始终使用复数。例如,如果您有一个包含销售订单行的字段‘lines’,您必须使用‘lines’而不是‘line’。

  • 如果您没有提供特定的路由处理程序来检索可用记录的列表,就不能期望客户端为您提供Odoo中特定记录的标识符(例如承运人的ID)。有时,客户端必须与Odoo共享特定记录的身份,以便能够执行针对该记录的特定操作(例如,每种支付获取器的支付处理方式不同)。在这种情况下,您必须提供特定属性,以便客户端和Odoo都能识别该记录。支付获取器上的“提供者”字段允许您在Odoo中识别特定记录。这种方法的优点是,客户端和Odoo都可以识别记录,而无需依赖于记录的ID(这可以确保如果记录的ID在Odoo中更改,例如在其他数据库上运行测试时,客户端不会崩溃)。

  • 始终为同一类型的对象使用相同的名称。例如,如果您有一个包含销售订单行列表的字段“lines”,则必须在所有其他JSON文档中使用相同的名称来表示同一类型的对象。

  • 以相同的方式管理JSON文档中对象之间的关系。默认情况下,您应该在JSON文档中返回相关对象的ID。但并非总是可能或方便,因此您还可以在JSON文档中返回相关对象。返回相关对象的主要优点是可以避免N+1问题。在JSON文档中返回相关对象的主要优点是可以避免调用以检索相关对象的额外操作。通过考虑每种方法的优缺点,您可以为您的情况选择最佳方法。一旦完成,您必须保持一致地管理同一对象的关系。

  • 在您的JSON文档中,将字段命名为与相应的Odoo模型中的字段相同的名称并不总是好主意。例如,在表示销售订单的文档中,您不应使用“order_line”来表示包含销售订单行列表的字段。“order_line”不仅容易混淆且不符合最佳实践,而且不具有自描述性。“lines”这个名称要好得多。

  • 保持一种防御性编程的方法。如果您提供了一个返回记录列表的路由处理程序,您必须确保列表的计算不会太长或耗尽服务器资源。例如,对于搜索路由处理程序,您必须确保默认情况下搜索仅限于合理的记录数。

  • 根据前一点,搜索处理程序必须始终使用合理的默认页面大小的分页机制。结果列表必须在一个包含记录总数和记录列表的JSON文档中。

  • 对于服务名称使用复数。例如,如果您提供了一个允许您管理销售订单的服务,您必须使用名称“sale_orders”而不是“sale_order”。

  • 等等。

我们可以写一本书来讲述在设计API时应遵循的最佳实践,但我们在这里就停止。这份清单是我们公司在ACSONE SA/NV的经验总结,并且随着时间的推移而不断发展。这是一套救援工具包,我们将其提供给开始设计API的新开发者。这个工具包必须伴随着阅读一些有用的资源链接,如REST指南。在技术层面上,fastapi文档提供了大量的有用信息以及许多示例。最后但并非最不重要的是,pydantic文档也非常有用。

杂项

搜索路由处理程序的开发

《odoo-addon-fastapi》模块提供了两段有用的代码,帮助您在编写搜索路由的处理程序时保持一致性。

  1. 一个依赖方法,用于以相同的方式指定所有搜索路由处理程序的分页参数:‘odoo.addons.fastapi.paging’

  2. 一个用于返回搜索路由处理程序结果的PagedCollection pydantic模型,该结果被包含在一个包含记录总数的JSON文档中。

import sys
if sys.version_info >= (3, 9):
    from typing import Annotated
else:
    from typing_extensions import Annotated
from pydantic import BaseModel

from odoo.api import Environment
from odoo.addons.fastapi.dependencies import paging, authenticated_partner_env
from odoo.addons.fastapi.schemas import PagedCollection, Paging

class SaleOrder(BaseModel):
    id: int
    name: str


@router.get(
    "/sale_orders",
    response_model=PagedCollection[SaleOrder],
    response_model_exclude_unset=True,
)
def get_sale_orders(
    paging: Annotated[Paging, Depends(paging)],
    env: Annotated[Environment, Depends(authenticated_partner_env)],
) -> PagedCollection[SaleOrder]:
    """Get the list of sale orders."""
    count = env["sale.order"].search_count([])
    orders = env["sale.order"].search([], limit=paging.limit, offset=paging.offset)
    return PagedCollection[SaleOrder](
        total=count,
        items=[SaleOrder.from_orm(order) for order in orders],
    )

错误处理的自定义

错误处理是fastapi与odoo集成设计中非常重要的话题。它必须确保错误消息被适当地返回给客户端,并且事务被适当地回滚。《fastapi》模块提供了一种注册自定义错误处理程序的方法。《odoo.addons.fastapi.error_handlers》模块提供了在创建《FastAPI》类的新实例时默认注册的默认错误处理程序。当在“fastapi.endpoint”模型中初始化应用程序时,会调用方法_get_app_exception_handlers来获取错误处理程序的字典。此方法设计为可以在自定义模块中重写,以提供自定义错误处理程序。您可以重写特定异常类的处理程序,或者为新的异常添加新的处理程序,甚至用您自己的处理程序替换所有处理程序。无论您做什么,您都必须确保事务被适当地回滚。

有些人可能会认为错误处理无法扩展,因为错误处理程序是全局方法,而不是在odoo模型中定义的。由于提供错误处理程序定义的方法是在“fastapi.endpoint”模型中定义的,所以这根本不是问题,您只需要换一种方式思考,而不是通过继承。

一个可能的解决方案是开发您自己的错误处理程序,以便能够处理错误并调用默认错误处理程序。

class MyCustomErrorHandler():
    def __init__(self, next_handler):
        self.next_handler = next_handler

    def __call__(self, request: Request, exc: Exception) -> JSONResponse:
        # do something with the error
        response = self.next_handler(request, exc)
        # do something with the response
        return response

使用此解决方案,您现在可以通过在您的自定义模块中重写方法_get_app_exception_handlers来注册您的自定义错误处理程序。

class FastapiEndpoint(models.Model):
    _inherit = "fastapi.endpoint"

    def _get_app_exception_handlers(
        self,
    ) -> Dict[
        Union[int, Type[Exception]],
        Callable[[Request, Exception], Union[Response, Awaitable[Response]]],
    ]:
        handlers = super()._get_app_exception_handlers()
        access_error_handler = handlers.get(odoo.exceptions.AccessError)
        handlers[odoo.exceptions.AccessError] = MyCustomErrorHandler(access_error_handler)
        return handlers

在先前的示例中,我们扩展了所有端点的‘AccessError’异常的处理程序。您可以为特定的应用程序做同样的事情,通过在注册您的自定义错误处理程序之前检查‘fastapi.endpoint’记录的‘app’字段。

FastAPI插件目录结构

当您开发一个新的插件以使用fastapi公开API时,遵循相同的目录结构和文件命名约定是一个好习惯。这有助于您轻松找到与API相关的文件,并有助于其他开发者理解您的代码。

以下是我们的推荐目录结构。它基于在开发fastapi应用程序时Python社区中使用的实践。

.
├── x_api
│   ├── data
│   │   ├── ... .xml
│   ├── demo
│   │   ├── ... .xml
│   ├── i18n
│   │   ├── ... .po
│   ├── models
│   │   ├── __init__.py
│   │   ├── fastapi_endpoint.py  # your app
│   │   └── ... .py
│   └── routers
│   │   ├── __init__.py
│   │   ├── items.py
│   │   └── ... .py
│   ├── schemas | schemas.py
│   │   ├── __init__.py
│   │   ├── my_model.py  # pydantic model
│   │   └── ... .py
│   ├── security
│   │   ├── ... .xml
│   ├── views
│   │   ├── ... .xml
│   ├── __init__.py
│   ├── __manifest__.py
│   ├── dependencies.py  # custom dependencies
│   ├── error_handlers.py  # custom error handlers
  • 《models》目录包含odoo模型。当您定义一个新应用程序时,就像其他插件一样,您将在此目录中添加继承自《fastapi.endpoint》模型的新模型。

  • 《routers》目录包含fastapi路由器。您将在此目录中添加您的新路由器。具有相同前缀的所有路由应分组在同一文件中。例如,所有以‘/items’开头的路由应定义在《items.py》文件中。此目录中的《__init__.py》文件用于导入目录中定义的所有路由器,并创建一个全局路由器,该路由器可以在应用程序中使用。例如,在您的《items.py》文件中,您将定义一个类似于这样的路由器:

    router = APIRouter(tags=["items"])
    
    router.get("/items", response_model=List[Item])
    def list_items():
        pass

    ‘__init__.py’ 文件中,您将导入路由器并将其添加到全局路由器或您的插件中。

    from fastapi import APIRouter
    
    from .items import router as items_router
    
    router = APIRouter()
    router.include_router(items_router)
  • ‘schemas.py’ 文件将用于定义 Pydantic 模型。对于具有许多模型的复杂 API,最好创建一个 ‘schemas’ 目录并将模型拆分到不同的文件中。该目录中的 ‘__init__.py’ 文件将用于导入目录中定义的所有模型。例如,在您的 ‘my_model.py’ 文件中,您将定义一个模型如下

    from pydantic import BaseModel
    
    class MyModel(BaseModel):
        name: str
        description: str = None

    ‘__init__.py’ 文件中,您将导入目录中的模型类。

    from .my_model import MyModel

    这将允许始终从 schemas 模块导入模型,无论模型是分散在不同文件中还是在 ‘schemas.py’ 文件中定义。

    from x_api_addon.schemas import MyModel
  • ‘dependencies.py’ 文件包含您将在路由器中使用的自定义依赖项。例如,您可以定义一个依赖项来检查用户的访问权限。

  • ‘error_handlers.py’ 文件包含您将在路由器中使用的自定义错误处理器。`odoo-addon-fastapi` 模块提供了常见 Odoo 异常的默认错误处理器。您可能不需要定义自己的错误处理器。但如果您需要这样做,您可以在该文件中定义它们。

接下来是什么?

`odoo-addon-fastapi` 模块仍处于开发初期。它将随着时间的推移而发展,以整合您的反馈并提供缺失的功能。现在轮到您尝试它并给出您的反馈了。

已知问题/路线图

路线图和 已知问题 可以在 GitHub 上找到。

`FastAPI` 模块提供了一种使用 WebSocket 的简单方法。不幸的是,这种支持目前尚不可用。挑战很大,因为 fastapi 的集成基于使用特定的中间件,该中间件将 Odoo 消耗的 WSGI 请求转换为 ASGI 请求。问题是是否也可以为 WebSocket 和流式传输大响应开发相同类型的桥接器。

错误跟踪器

错误在 GitHub Issues 上跟踪。如果遇到问题,请检查是否已报告您的问题。如果您是第一个发现它的人,请帮助我们通过提供详细且受欢迎的 反馈 来解决问题。

请不要直接联系贡献者以获取支持或技术问题的帮助。

鸣谢

作者

  • ACSONE SA/NV

贡献者

维护者

此模块由 OCA 维护。

Odoo Community Association

OCA,即 Odoo 社区协会,是一个非营利组织,其使命是支持 Odoo 功能的协作开发并促进其广泛使用。

当前 维护者

lmignon

此模块是 GitHub 上 OCA/rest-framework 项目的一部分。

欢迎您贡献力量。要了解如何贡献力量,请访问 https://odoo-community.org/page/Contribute

项目详情


下载文件

下载适用于您平台的文件。如果您不确定选择哪个,请了解更多关于 安装包 的信息。

源代码分布

本发行版没有可用的源代码分布文件。请参阅有关 生成发行版存档 的教程。

构建的分布

odoo14_addon_fastapi-14.0.1.0.0-py3-none-any.whl (171.9 kB 查看散列值)

上传时间 Python 3

由以下支持