如何使用 sqlalchemy 和 Pydantic 通过 FastPI 构建 postgresql 数据库

Question

我想使用 sqlalchemy 和 Pydantic 构建一个带有 FastPI 的 postgresql 数据库。

在输入字典my_car_dict（这是一个嵌套字典）中存储数据库的所有数据。我想使用 Pydantic 创建嵌套字典的数据库。但是，我不确定我是否在从 BaseModel 继承的类中正确定义了嵌套字典结构。我在下面提供了我的例子。

我按照 https://fastapi.tiangolo.com/id/tutorial/sql-databases/ 中的示例进行操作 为了创建数据库，我的文件夹结构如下所示：

sql_app/
    __init__.py
    crud.py
    database.py
    main.py
    models.py
    schemas.py

我的crud.py文件：

from sqlalchemy.orm import Session

from . import models, schemas

def get_car_properties(db: Session, car_id: int, prop_id: int):
    return db.query(models.Car).filter(models.CarDescription.carid == car_id, models.CarProperties.propid == prop_id).first()

我的database.py*文件：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

我的main.py文件：

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
        
@app.get("/Cars/{car_id}{prop_id}", response_model=schemas.Car)
def read_car(car_id: int, prop_id: int, db: Session = Depends(get_db)):
    db_car = crud.get_car_properties(db, car_id=car_id, prop_id=prop_id)
    if db_car is None:
        raise HTTPException(status_code=404, detail="Car not found")
    return db_car

我的models.py文件：

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base

class Cars(Base):
    __tablename__ = "cars"

    carid = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    cartype = Column(String)
    comment = Column(String)
    speedrange = Column(String)
    
    properties = relationship("Properties", back_populates="owner")

class Properties(Base):
    __tablename__ = "properties"
        
    propid = Column(Integer, primary_key=True, index=True)
    mass = Column(String)
    unit = Column(String)
    hp = Column(String)
    carmodels = Column(String)
    price = Column(String)
    source = Column(String)
    fuel_consumption = Column(String)
    
    owner = relationship("Cars", back_populates="properties") 

最后是我的 schemas.py 文件：

from typing import Union, List, Optional, Dict
from pydantic import BaseModel

my_car_dict = {
        "1": {
            "carid": "1",
            "name": "Porsche",
            "classification": {
                "cartype": "911",
                "comment": "sportscar",
                "speedrange": "[0:182]",
                "properties": {
                    "3": {
                        "carmodels": ["Turbo, GT, Carrera"],
                        "unit": "-",
                        "propid": "3",
                    },
                    "6": {
                        "hp": "500",
                        "unit": "ft-lb/s",
                        "propid": "3",
                    },
                    "8": {
                        "price": "150000",
                        "source": "4",
                        "propid": "8",
                        "unit": "$"
                        }
                    }
                }
            },
        "2": {        
            "carid": "2",
            "name": "Volkswagen",
            "classification": {
                "cartype": "Golf",
                "comment": "hatchback",
                "speedrange": "[0:142]",
                "properties": {
                    "3": {
                        "mass": "3527",
                        "unit": "lbs",
                        "propid": "3",
                    },
                    "6": {
                        "hp": "150",
                        "unit": "ft-lb/s",
                        "propid": "3",
                    },
                    "8": {
                        "price": "40000",
                        "source": "4",
                        "propid": "8",
                        "unit": "$"
                    },
                    "13": {
                        "fuel_consumption": "6",
                        "propid": "25",
                        "unit": "l/100km"
                    }
                }
            }
        }
}

class CarProperties(BaseModel):
    propid: int
    mass: float
    unit: str
    hp: float
    carmodels: List[str]
    price: float
    source: int
    fuel_consumption: float

class CarClassification(BaseModel):
    cartype: str
    comment: str
    speedrange: List[float]
    properties: Dict[int, Dict[str, CarProperties]]
    
class CarDescription(BaseModel):
    carid: int
    name: str
    classification: Dict[str, CarClassification]

class Car(BaseModel):
    for key in my_car_dict.keys():
        key: Dict[str, Dict[str, CarDescription]]    
        
    class Config:
        orm_mode = True 

但是，当尝试运行它时：

uvicorn sql_app.main:app --reload

我收到此错误：

(base) PS C:\Users\XXXXXXX\desktop\sql_app> uvicorn sql_app.main:app --reload
←[32mINFO←[0m:     Will watch for changes in these directories: ['C:\\Users\XXXXXXX\desktop\\sql_app']
←[32mINFO←[0m:     Uvicorn running on ←[1mhttp://127.0.0.1:8000←[0m (Press CTRL+C to quit)
←[32mINFO←[0m:     Started reloader process [←[36m←[1m29524←[0m] using ←[36m←[1mStatReload←[0m
Process SpawnProcess-1:
Traceback (most recent call last):
  File "C:\Tools\DEV\Anaconda3_2023\Lib\multiprocessing\process.py", line 314, in _bootstrap
    self.run()
  File "C:\Tools\DEV\Anaconda3_2023\Lib\multiprocessing\process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Tools\DEV\Anaconda3_2023\Lib\site-packages\uvicorn\_subprocess.py", line 76, in subprocess_started
    target(sockets=sockets)
  File "C:\Tools\DEV\Anaconda3_2023\Lib\site-packages\uvicorn\server.py", line 61, in run
    return asyncio.run(self.serve(sockets=sockets))
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Tools\DEV\Anaconda3_2023\Lib\asyncio\runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "C:\Tools\DEV\Anaconda3_2023\Lib\asyncio\runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Tools\DEV\Anaconda3_2023\Lib\asyncio\base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "C:\Tools\DEV\Anaconda3_2023\Lib\site-packages\uvicorn\server.py", line 68, in serve
    config.load()
  File "C:\Tools\DEV\Anaconda3_2023\Lib\site-packages\uvicorn\config.py", line 467, in load
    self.loaded_app = import_from_string(self.app)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Tools\DEV\Anaconda3_2023\Lib\site-packages\uvicorn\importer.py", line 24, in import_from_string
    raise exc from None
  File "C:\Tools\DEV\Anaconda3_2023\Lib\site-packages\uvicorn\importer.py", line 21, in import_from_string
    module = importlib.import_module(module_str)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Tools\DEV\Anaconda3_2023\Lib\importlib\__init__.py", line 126, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<frozen importlib._bootstrap>", line 1206, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1178, in _find_and_load
  File "<frozen importlib._bootstrap>", line 1128, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "<frozen importlib._bootstrap>", line 1206, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1178, in _find_and_load
  File "<frozen importlib._bootstrap>", line 1142, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'sql_app'

有人知道我做错了什么吗？

Answer 1

似乎是进口问题。您可能应该尝试依赖源目录（绝对导入），在您的情况下是 sql_app。 所以进口是：

from sql_app.crud import get_car_properties
from sql_app.database import SessionLocal

等等。当我使用 dockerized 环境时，我遇到了同样的问题。