Сьогодні ми зробимо ще один крок уперед і заглибимось у те, як інтегрувати SQLAlchemy з FastAPI для виконання CRUD-операцій через API.
Готові? Наливайте каву, починаємо!
Як тільки наші моделі даних налаштовані в SQLAlchemy, нам потрібно надати інтерфейс для взаємодії з цими даними.
І тут FastAPI показує себе з найкращого боку: він дозволяє легко створювати маршрути (endpoints) для виконання CRUD операцій.
Приклад моделі даних
Почнемо з базової моделі даних, яку ми вже створювали раніше.
Припустімо, у нас є таблиця User, що містить інформацію про користувачів.
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
Налаштування підключення та сесії
Для початку нам потрібно налаштувати підключення до бази даних. Це серце взаємодії з базою даних через FastAPI.
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db" # Використовуємо SQLite для прикладу
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Створюємо таблиці
Base.metadata.create_all(bind=engine)
Ми знову використовуємо SQLite для простоти, але ти можеш легко замінити це на PostgreSQL або будь-яку іншу реляційну базу даних.
Створення маршрутів (endpoints)
Тепер, коли підключення налаштовано, створимо API-маршрути для виконання CRUD-операцій. Почнемо з простого FastAPI-додатку.
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
app = FastAPI()
# Залежність для отримання сесії бази даних
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Функції для операцій
Створимо функції для кожної CRUD-операції: створення, читання, оновлення і видалення.
Створення користувача
from pydantic import BaseModel
class UserCreate(BaseModel):
name: str
email: str
@app.post("/users/", response_model=UserCreate)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(name=user.name, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
Коли користувач надсилає запит на створення (POST), ми додаємо новий об'єкт в базу даних, фіксуємо зміни за допомогою db.commit() і повертаємо створений об'єкт.
Читання користувачів
from typing import List
@app.get("/users/", response_model=List[UserCreate])
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
users = db.query(User).offset(skip).limit(limit).all()
return users
Ми використовуємо skip і limit як параметри запиту, щоб реалізувати пагінацію. Це корисно при великих об'ємах даних.
Оновлення користувача
@app.put("/users/{user_id}", response_model=UserCreate)
def update_user(user_id: int, updated_user: UserCreate, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
raise HTTPException(status_code=404, detail="User not found")
db_user.name = updated_user.name
db_user.email = updated_user.email
db.commit()
db.refresh(db_user)
return db_user
Ми шукаємо користувача за id, оновлюємо його дані, фіксуємо зміни і повертаємо оновлений об'єкт.
Видалення користувача
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.id == user_id).first()
if not db_user:
raise HTTPException(status_code=404, detail="User not found")
db.delete(db_user)
db.commit()
return {"message": "User deleted successfully"}
Тут ми шукаємо користувача, видаляємо його і фіксуємо зміни в базі даних. Якщо користувача не знайдено, повертаємо помилку 404.
Приклад виконання асинхронних операцій
FastAPI підтримує асинхронні функції, але важливо розуміти, що поки SQLAlchemy (у своєму стандартному вигляді) працює синхронно.
Для асинхронних операцій нам знадобиться використовувати бібліотеку asyncpg разом з databases.
Встановлення необхідних бібліотек
pip install databases asyncpg
Асинхронне налаштування підключення
Використовуємо бібліотеку databases для асинхронної взаємодії.
from databases import Database
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/testdb"
database = Database(DATABASE_URL)
@app.on_event("startup")
async def startup():
await database.connect()
@app.on_event("shutdown")
async def shutdown():
await database.disconnect()
Асинхронні CRUD-операції реалізуються практично так само, як і синхронні. Наприклад:
Створення запису
@app.post("/async_users/")
async def create_async_user(user: UserCreate):
query = "INSERT INTO users(name, email) VALUES (:name, :email)"
await database.execute(query=query, values={"name": user.name, "email": user.email})
return {"message": "User created!"}
Читання записів
@app.get("/async_users/")
async def read_async_users():
query = "SELECT * FROM users"
return await database.fetch_all(query=query)
Асинхронні операції особливо корисні при високих навантаженнях або при взаємодії з іншими API.
Тестування CRUD операцій через API
Для тестування ми можемо використовувати інструменти на кшталт Postman або curl.
Але автоматичне тестування — це must-have для будь-якого серйозного проєкту.
Давайте створимо тести з використанням Pytest і TestClient.
FastAPI надає TestClient для тестування.
Встановлення необхідних бібліотек
pip install pytest httpx
Приклад тесту
from fastapi.testclient import TestClient
client = TestClient(app)
def test_create_user():
response = client.post("/users/", json={"name": "John", "email": "john@example.com"})
assert response.status_code == 200
assert response.json()["email"] == "john@example.com"
Цей тест перевіряє маршрут POST для створення користувача. Аналогічно можна тестувати читання, оновлення і видалення.
Тепер наш додаток повноцінно взаємодіє з базою даних через SQLAlchemy і FastAPI.
Ми реалізували основні CRUD операції, вивчили асинхронні можливості та автоматичне тестування.
Це не тільки корисно для реальних проєктів, але й вразить будь-якого інтерв'юера на співбесіді! 🚀
ПЕРЕЙДІТЬ В ПОВНУ ВЕРСІЮ