定义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

这个导入错误是由以下过程引起的:

1. supplier_schema.py被定义
2. supplier_schema模块从schemas.category_schema导入Category
3. 由于schemas.category_schema尚未定义，所以现在执行。
4. 但等等!category_schema需要从supplier_schema导入Supplier。
5. supplier_schema不能在没有从category_schema导入的情况下继续执行，反之亦然。
6. 这就是循环导入。

如何解决循环导入?

我将处理这个特定的情况。
有关更一般的用例，请阅读: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解决了这个问题

你可以参考这个和这个来获得更多的信息。

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