FastAPI ve MongoDB için Odmantic ile Tip Güvenli CRUD

· Burak Aslan · 4 dk okuma · Python , MongoDB , FastAPI

Selamlar, bu yazıda FastAPI ile MongoDB konuşurken araya Odmantic'i koyduğumuzda işlerin ne kadar sadeleştiğine bakacağız. Motor ile dökme dokümanlar üzerinde sözlük gezindirmek bir noktadan sonra bana hep can sıkıcı geldi; her endpoint'te elle dict doğrulamak, _id'yi str'ye çevirmek, eksik alanı yakalamaya çalışmak. Odmantic Pydantic'in tip sistemini Motor'un üstüne kuruyor ve bu üç parça birden FastAPI'nın dependency injection mekanizmasıyla kucaklaşıyor. Lafı uzatmadan başlayalım.

Odmantic nedir?

Odmantic, MongoDB için yazılmış async bir ODM (object document mapper). Schema tanımı için Pydantic modellerini, sürücü olarak da Motor'u kullanıyor. Yani siz class Product(Model) yazıyorsunuz, Odmantic bu sınıfı hem MongoDB tarafında bir koleksiyona, hem de FastAPI tarafında otomatik bir request/response şemasına bağlıyor. OpenAPI dokümantasyonu bedavaya geliyor, çünkü Pydantic zaten FastAPI'nın doğal dili.

Kurulum tek satır:

pip install odmantic motor fastapi uvicorn

Model tanımı

Bir model odmantic.Model'den türüyor. Aşağıda küçük bir ürün modeli tanımlıyoruz:

from typing import List
from datetime import datetime
from odmantic import Model, Field

class Product(Model):
    name:      str
    price:     float
    category:  str
    tags:      List[str] = []
    inStock:   bool = True
    createdAt: datetime = Field(default_factory=datetime.utcnow)

    model_config = {'collection': 'products'}

Burada koleksiyon adını model_config ile sabitledik. Vermezseniz Odmantic sınıf adından türetir; ben açıkça yazmayı tercih ediyorum çünkü prod'da bir gün biri sınıfı yeniden adlandırır ve koleksiyon ortadan kaybolur.

AIOEngine: motorun arkasındaki motor

Odmantic'in Motor'u sardığı yere AIOEngine deniyor. Bütün CRUD bu nesne üzerinden geçiyor:

from motor.motor_asyncio import AsyncIOMotorClient
from odmantic import AIOEngine

client = AsyncIOMotorClient('mongodb://localhost:27017/')
engine = AIOEngine(client=client, database='shop')

Engine'i uygulama yaşam döngüsüne bağlamak şart; her istekte yeni client açmak hem yavaş hem de connection pool'u boşa harcar.

CRUD'un dört temel hareketi

Tek bir korutin içinde dört işlemi de görelim:

import asyncio
from odmantic import AIOEngine
from motor.motor_asyncio import AsyncIOMotorClient

async def main():
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    engine = AIOEngine(client=client, database='shop')

    # Olustur
    product = Product(name='Laptop', price=999.99, category='electronics')
    await engine.save(product)
    print('Kayit edildi:', product.id)

    # Tek kayit getir
    bulunan = await engine.find_one(Product, Product.name == 'Laptop')

    # Cogul kayit getir, fiyata gore sirali
    ucuzlar = await engine.find(Product, Product.price < 100, sort=Product.price)

    # Guncelle
    bulunan.price = 899.99
    await engine.save(bulunan)

    # Sil
    await engine.delete(bulunan)

asyncio.run(main())

Dikkat ederseniz engine.save() hem insert hem update için yeterli. Belge id taşıyorsa update, taşımıyorsa insert oluyor. Sorgu filtreleri de düz Python operatörleriyle yazılıyor: Product.price < 100 gibi ifadeler arka planda MongoDB sorgusuna çevriliyor. Bu kısım benim en sevdiğim taraf.

FastAPI tarafına bağlamak

Şimdi engine'i FastAPI'nın lifespan mekanizmasına yerleştirelim ve birkaç endpoint açalım:

from typing import List
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Depends
from motor.motor_asyncio import AsyncIOMotorClient
from odmantic import AIOEngine, ObjectId

engine: AIOEngine | None = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global engine
    client = AsyncIOMotorClient('mongodb://localhost:27017/')
    engine = AIOEngine(client=client, database='shop')
    yield
    client.close()

app = FastAPI(lifespan=lifespan)

def get_engine() -> AIOEngine:
    assert engine is not None
    return engine

@app.get('/products', response_model=List[Product])
async def list_products(eng: AIOEngine = Depends(get_engine)):
    return await eng.find(Product, sort=Product.price)

@app.post('/products', response_model=Product)
async def create_product(product: Product, eng: AIOEngine = Depends(get_engine)):
    return await eng.save(product)

@app.get('/products/{product_id}', response_model=Product)
async def get_product(product_id: ObjectId, eng: AIOEngine = Depends(get_engine)):
    bulunan = await eng.find_one(Product, Product.id == product_id)
    if not bulunan:
        raise HTTPException(status_code=404, detail='Urun bulunamadi')
    return bulunan

Bence buradaki en güzel detay product_id: ObjectId parametresi. FastAPI gelen string'i otomatik olarak MongoDB'nin ObjectId tipine çeviriyor; geçersiz id gelirse 422 dönüyor. Eskiden elle try/except yazdığımız bu doğrulama tek satıra inmiş oluyor.

Sık karşılaşılan tuzaklar

  • Engine'i modül seviyesinde init etmek: lifespan dışında oluşturursanız test ortamında event loop çakışması yaşarsınız. Mutlaka lifespan içinde kurun.
  • Product'ı hem request hem DB modeli olarak kullanmak: Küçük projede sorun yok, ama büyüdükçe ProductIn ve ProductOut ayırmak gerekebilir; aksi halde istemci id gönderebilir hâle gelir.
  • find() sonucuna ezbere güvenmek: Cursor değil, list dönüyor; çok büyük koleksiyonda limit ve skip kullanmadan çekerseniz belleği yorarsınız.
  • datetime.utcnow deprecation uyarısı: Python 3.12+ için datetime.now(timezone.utc) tercih edin; Odmantic her ikisini de kabul ediyor.

Kapanış

Odmantic, Motor'un hız ve esnekliğini Pydantic'in disipliniyle birleştiriyor; FastAPI ile yan yana koyduğunuzda CRUD katmanı neredeyse kendiliğinden ortaya çıkıyor. Bence küçük ve orta ölçekli MongoDB servisleri için Beanie'ye göre daha sade, ham Motor'a göre çok daha güvenli bir orta yol. Umarım faydalı olur, bir sonraki yazıda görüşmek üzere.