定义 Fastapi Pydantic 多对多关系

Question

在 pydantic 模型中定义多对多关系而不出现循环导入错误的正确方法是什么。我有两个文件 supplier_schema.py 和 category_schema.py.

supplier_schema.py

from pydantic import BaseModel, HttpUrl, EmailStr
from typing import Optional, List
from datetime import datetime
from schemas.category_schema import Category
from schemas.membership_schema import Membership


class SupplierBase(BaseModel):
    contact_person: str
    email_address: EmailStr
    company_name: str
    company_email: str
    company_logo: str
    country: str
    state_province: str
    city_area: Optional[str] = None
    location: str
    phone: str
    fax: Optional[str] = None
    tagline: str
    company_bio: Optional[str]
    year_established: int
    employees_count: str
    certificates: Optional[List[str]] = None
    cover_image: Optional[HttpUrl] = None
    gallery: Optional[List[HttpUrl]] = None
    company_license: Optional[HttpUrl] = None
    company_website: Optional[HttpUrl] = None
    whatsapp_number: Optional[str] = None
    facebook_url: Optional[HttpUrl] = None
    twitter_url: Optional[HttpUrl] = None
    linkedin_url: Optional[HttpUrl] = None
    pinterest_url: Optional[HttpUrl] = None
    instagram_url: Optional[HttpUrl] = None
    youtube_url: Optional[HttpUrl] = None
    annual_revenue: Optional[str] = None
    challenges: List[str]
    status: str = None

    class Config:
        orm_mode = True


class SupplierCreate(SupplierBase):
    password: str
    membership_id: int
    categories_id: Optional[List[int]] = []


class Supplier(SupplierBase):
    id: int
    created_at: datetime
    updated_at: datetime
    membership: Membership
    categories: List[Category]

category_schema.py

from pydantic import BaseModel, HttpUrl, EmailStr
from typing import Optional, List
from datetime import datetime
from schemas.category_schema import Category
from schemas.membership_schema import Membership


class SupplierBase(BaseModel):
    contact_person: str
    email_address: EmailStr
    company_name: str
    company_email: str
    company_logo: str
    country: str
    state_province: str
    city_area: Optional[str] = None
    location: str
    phone: str
    fax: Optional[str] = None
    tagline: str
    company_bio: Optional[str]
    year_established: int
    employees_count: str
    certificates: Optional[List[str]] = None
    cover_image: Optional[HttpUrl] = None
    gallery: Optional[List[HttpUrl]] = None
    company_license: Optional[HttpUrl] = None
    company_website: Optional[HttpUrl] = None
    whatsapp_number: Optional[str] = None
    facebook_url: Optional[HttpUrl] = None
    twitter_url: Optional[HttpUrl] = None
    linkedin_url: Optional[HttpUrl] = None
    pinterest_url: Optional[HttpUrl] = None
    instagram_url: Optional[HttpUrl] = None
    youtube_url: Optional[HttpUrl] = None
    annual_revenue: Optional[str] = None
    challenges: List[str]
    status: str = None

    class Config:
        orm_mode = True


class SupplierCreate(SupplierBase):
    password: str
    membership_id: int
    categories_id: Optional[List[int]] = []


class Supplier(SupplierBase):
    id: int
    created_at: datetime
    updated_at: datetime
    membership: Membership
    categories: List[Category]

但是我得到这个错误

Traceback (most recent call last):
  File "/usr/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/usr/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/home/phat/projects/crdle2-api/venv/lib/python3.9/site-packages/uvicorn/subprocess.py", line 61, in subprocess_started
    target(sockets=sockets)
  File "/home/phat/projects/crdle2-api/venv/lib/python3.9/site-packages/uvicorn/server.py", line 49, in run
    loop.run_until_complete(self.serve(sockets=sockets))
  File "uvloop/loop.pyx", line 1501, in uvloop.loop.Loop.run_until_complete
  File "/home/phat/projects/crdle2-api/venv/lib/python3.9/site-packages/uvicorn/server.py", line 56, in serve
    config.load()
  File "/home/phat/projects/crdle2-api/venv/lib/python3.9/site-packages/uvicorn/config.py", line 308, in load
    self.loaded_app = import_from_string(self.app)
  File "/home/phat/projects/crdle2-api/venv/lib/python3.9/site-packages/uvicorn/importer.py", line 23, in import_from_string
    raise exc from None
  File "/home/phat/projects/crdle2-api/venv/lib/python3.9/site-packages/uvicorn/importer.py", line 20, in import_from_string
    module = importlib.import_module(module_str)
  File "/usr/lib/python3.9/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1030, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
  File "<frozen importlib._bootstrap>", line 986, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 680, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 850, in exec_module
  File "<frozen importlib._bootstrap>", line 228, in _call_with_frames_removed
  File "/home/phat/projects/crdle2-api/./main.py", line 3, in <module>
    from routers import supplier_router, membership_router, category_router
  File "/home/phat/projects/crdle2-api/./routers/supplier_router.py", line 4, in <module>
    from schemas.supplier_schema import Supplier, SupplierCreate
  File "/home/phat/projects/crdle2-api/./schemas/supplier_schema.py", line 4, in <module>
    from schemas.category_schema import Category
  File "/home/phat/projects/crdle2-api/./schemas/category_schema.py", line 4, in <module>
    from schemas.supplier_schema import Supplier
ImportError: cannot import name 'Supplier' from partially initialized module 'schemas.supplier_schema' (most likely due to a circular import) (/home/phat/projects/crdle2-api/./schemas/supplier_schema.py)

我做错了什么？我该如何解决这个问题？

Answer 1

此导入错误是由于以下过程造成的：

supplier_schema.py定义
supplier_schema 模块从 schemas.category_schema

Category

由于schemas.category_schema还没有定义，现在执行
但是等等！ category_schema 需要从 supplier_schema.

Supplier

category_schema

supplier_schema 无法继续执行，反之亦然。
这被称为 circular import。

如何解决循环导入？

I'll be addressing this specific case.
For more general use cases, read: Circular import dependency in Python

选项 1： 将所有内容合并到一个模块中。

这是最快最简单的解决方案。
一旦您的 ORM 代码增加到许多模型，就会变得非常失控。

选项 2： 在根模块的初始化中导入模型并更改导入语言：

稳健的解决方案
需要更仔细地规划和设计您的项目
需要测试（额外的进口会造成破损）

执行方法如下：
确保您的 schemas 模块包含一个 init 文件：

-- schemas
|-- __init__.py
|-- category_schema.py
|-- membership_schema.py

您的 __init__.py 文件应包含以下内容：

from . import category_schema
from . import membership_schema

现在，您的导入应修改为以下内容：

# Old way: ⬇
# from schemas.category_schema import Category

# New way: ⬇
import schemas

...

class Supplier(SupplierBase):    
    categories: List[schemas.category_schema.Category]
...

重复其他子模块和导入。

关于选择解决方案的忠告：

循环导入是使用 ORM 时非常常见的错误。遇到一个意味着您已经大大扩展了应用程序的属性——恭喜！根据您的目标为您的项目选择最佳解决方案。

如果您的代码是概念验证并且您需要快速行动，我建议您选择选项 1。（快速行动并打破常规。）
如果你的代码是一个更健壮的项目的一部分，即使你处于早期阶段，一定要选择选项 2。（今天解决明天的问题。）

Answer 2

所以，我通过使用注释 from __future__ import annotations 和 pydantic 的 update_forward_refs

解决了这个问题

您可以参考this and 了解更多信息。

supplier_schema.py

from __future__ import annotations
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime
# import schemas


class SupplierBase(BaseModel):
    company_email: Optional[str] = None
    logo: Optional[str] = None
    state_province: str = None
    city_area: Optional[str] = None
    location: Optional[str] = None
    company_phone: str = None
    fax: Optional[str] = None
    tagline: Optional[str] = None
    company_bio: Optional[str] = None
    postal_code: Optional[str] = None
    year_established: str = None
    employees_count: Optional[str] = None
    certificates: Optional[List[dict]] = []
    cover_image: Optional[str] = None
    gallery: Optional[List[dict]] = None
    license: Optional[str] = None
    website: Optional[str] = None
    whatsapp_number: Optional[str] = None
    facebook_url: Optional[str] = None
    twitter_url: Optional[str] = None
    linkedin_url: Optional[str] = None
    pinterest_url: Optional[str] = None
    instagram_url: Optional[str] = None
    youtube_url: Optional[str] = None
    annual_revenue: Optional[str] = None
    payment_methods: Optional[List[dict]] = None
    challenges: Optional[List[str]] = []
    status: str = None
    user_id: int

    class Config:
        orm_mode = True


class SupplierCreate(SupplierBase):
    pass


class Supplier(SupplierBase):
    id: int
    categories: Optional[List[Category]] = []
    created_at: datetime
    updated_at: datetime


from .category_schema import Category  # nopep8
Category.update_forward_refs()

category_schema.py

from __future__ import annotations
from typing import List
from pydantic import BaseModel
from datetime import datetime


class CategoryBase(BaseModel):
    label: str

    class Config:
        orm_mode = True


class CategoryCreate(CategoryBase):
    pass


class Category(CategoryBase):
    id: int
    created_at: datetime
    updated_at: datetime
    suppliers: List[Supplier]


from .supplier_schema import Supplier  # nopep8
Category.update_forward_refs()


class CategorySkeleton(CategoryBase):
    pass


class CategoryPatch(CategoryBase):
    pass

定义 Fastapi Pydantic 多对多关系

Defining Fastapi Pydantic many to Many relationships

python

sqlalchemy

pydantic

fastapi

如何解决循环导入？

关于选择解决方案的忠告：