Доброго времени суток! Я понимаю, что вам нужно создать универсальный класс маршрутов, который будет работать с разными моделями в FastAPI. Это абсолютно возможно, и я рад помочь вам с этим.
Вот пример реализации такого подхода. Мы создадим базовый класс для CRUD-операций, который будет принимать модель и схемы Pydantic. Также будет возможность наследования для переопределения отдельных маршрутов.
```python
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
from typing import TypeVar, Generic, List, Type
# Импортируйте свои модели и зависимости для доступа к базе данных
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
# Пример модели SQLAlchemy
class Item(Base):
__tablename__ = 'items'
id: int
name: str
# Создание базы данных
DATABASE_URL = "sqlite:///./test.db" # Замените на URL вашей БД
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Общее API
app = FastAPI()
ModelType = TypeVar("ModelType", bound=BaseModel)
DBModelType = TypeVar("DBModelType", bound=Base)
class CRUDRouter(Generic[ModelType, DBModelType]):
def __init__(self, model: Type[DBModelType], schema: Type[ModelType]):
self.model = model
self.schema = schema
def get_all(self):
"""Получить все объекты"""
db: Session = Depends(get_db)
items = db.query(self.model).all()
return items
def get(self, item_id: int):
"""Получить объект по ID"""
db: Session = Depends(get_db)
item = db.query(self.model).filter(self.model.id == item_id).first()
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return item
def create(self, item: schema):
"""Создать новый объект"""
db: Session = Depends(get_db)
db_item = self.model(**item.dict())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
def update(self, item_id: int, item: schema):
"""Обновить объект по ID"""
db: Session = Depends(get_db)
db_item = db.query(self.model).filter(self.model.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
for key, value in item.dict().items():
setattr(db_item, key, value)
db.commit()
return db_item
def delete(self, item_id: int):
"""Удалить объект по ID"""
db: Session = Depends(get_db)
db_item = db.query(self.model).filter(self.model.id == item_id).first()
if not db_item:
raise HTTPException(status_code=404, detail="Item not found")
db.delete(db_item)
db.commit()
return {"detail": "Item deleted"}
# Зависимость для получения сессии базы данных
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Схема для Pydantic
class ItemSchema(BaseModel):
name: str
# Создайте маршруты для модели Item, используя CRUDRouter
crud_item = CRUDRouter(Item, ItemSchema)
@app.get("/items/", response_model=List[ItemSchema])
def read_items():
return crud_item.get_all()
@app.get("/items/{item_id}", response_model=ItemSchema)
def read_item(item_id: int):
return crud_item.get(item_id)
@app.post("/items/", response_model=ItemSchema)
def create_item(item: ItemSchema):
return crud_item.create(item)
@app.put("/items/{item_id}", response_model=ItemSchema)
def update_item(item_id: int, item: ItemSchema):
return crud_item.update(item_id, item)
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
return crud_item.delete(item_id)
```
### Объяснение
1. **CRUDRouter**: Мы создали обобщённый класс `CRUDRouter`, который принимает модель и схему Pydantic. Внутри класса мы определили методы для всех CRUD операций.
2. **Зависимости базы данных**: Мы используем асинхронные зависимости FastAPI для получения сессии базы данных.
3. **Маршруты**: Теперь вы