Блог

структура

 .
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py
 

main.py

 from typing import List

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.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_name(db, name=user.name)
    if db_user:
        raise HTTPException(status_code=400, detail="Name already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/messages/", response_model=schemas.Message)
def create_message_for_user(
    user_id: int, message: schemas.MessageCreate, db: Session = Depends(get_db)
):
    return crud.create_user_message(db=db, message=message, user_id=user_id)


@app.get("/messages/", response_model=List[schemas.MessageWithUser])
def read_messages(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    messages = crud.get_messages(db, skip=skip, limit=limit)
    return messages
 

crud.py

 from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_name(db: Session, name: str):
    return db.query(models.User).filter(models.User.name == name).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    db_user = models.User(name=user.name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_messages(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Message).offset(skip).limit(limit).all()


def create_user_message(db: Session, message: schemas.MessageCreate, user_id: int):
    db_item = models.Message(**message.dict(), user_id=user_id)
    db.add(db_item)
    db.commit()
db.refresh(db_item)
return db_item
 

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()
 

models.py

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

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, unique=True, nullable=False)

    messages = relationship("Message", back_populates="user")


class Message(Base):
    __tablename__ = "messages"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True, unique=True, nullable=False)
    message = Column(String, index=False)
    user_id = Column(Integer, ForeignKey("users.id"))

    user = relationship("User", back_populates="messages")
 

schemas.py

 from typing import List

from pydantic import BaseModel


class MessageBase(BaseModel):
    title: str
    message: str


class MessageCreate(MessageBase):
    pass


class Message(MessageBase):
    id: int
    user_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    name: str


class UserCreate(UserBase):
    pass


class User(UserBase):
    id: int
    messages: List[Message] = []

    class Config:
        orm_mode = True


class UserIdentifier(UserBase):
    id: int
    name: str


class MessageWithUser(MessageBase):
    id: int
    title: str
    message: str
    user: UserIdentifier
 

Запустите cmd:

 uvicorn sql_app.main:app --reload
 

Перейдите к: 127.0.0.1:8000/docs

Добавление пользователей и сообщений для пользователей

Попробуйте прочитать сообщения

Результат примера — правильный формат, но он не дает ответа, только ошибка 500 и следующее

Ошибка:

 pydantic.error_wrappers.ValidationError: 4 validation errors for MessageWithUser
response -> 0
  value is not a valid dict (type=type_error.dict)
response -> 1
  value is not a valid dict (type=type_error.dict)
response -> 2
  value is not a valid dict (type=type_error.dict)
response -> 3
  value is not a valid dict (type=type_error.dict)
 

Ожидаемый результат — это то, что показано в примере вывода страницы документов, при этом идентификатор пользователя добавляется в сообщение, но не какие-либо другие поля от пользователя, если таковые были, в данном случае нет

Я не думаю, что можно привести объект SQLAlchemy с зависимостями к модели pydantic из коробки.

@tiagolo (создатель FastAPI) создал этот метод pydantic-sqlalchemy, который, кажется, делает то, что вы хотите: https://github.com/tiangolo/pydantic-sqlalchemy

В противном случае, я думаю, вы можете создать список MessageWithUser вручную и вернуть его.