SQLAlchemy 2.0'ı FastAPI ile kullanmaya yönelik Kalıplar ve Uygulamalar

Django ve Flask birçok Python mühendisinin ilk tercihi olmaya devam ederken, FastAPI zaten inkar edilemez derecede güvenilir bir seçim olarak kabul edildi. Geliştiriciye arka uç uygulamaları oluşturma konusunda sonsuz olanaklar sunan oldukça esnek, iyi optimize edilmiş, yapılandırılmış bir çerçevedir.

Veritabanlarıyla çalışmak çoğu arka uç uygulamasının önemli bir yönüdür. Sonuç olarak ORM, arka uç kodunda kritik bir rol oynar. Ancak Django'nun aksine FastAPI'de yerleşik bir ORM yoktur. Uygun bir kütüphane seçmek ve onu kod tabanına entegre etmek tamamen geliştiricinin sorumluluğundadır.

Python mühendisleri SQLAlchemy'nin mevcut en popüler ORM olduğunu düşünüyor. 2006 yılından bu yana kullanılan ve binlerce proje tarafından benimsenen efsane bir kütüphanedir. 2023 yılında 2.0 sürümüne büyük bir güncelleme geldi. FastAPI'ye benzer şekilde SQLAlchemy, geliştiricilere onları belirli bir şekilde kullanmaya zorlamadan güçlü özellikler ve yardımcı programlar sağlar. Temel olarak, geliştiricilerin onu uygun gördükleri şekilde kullanmalarına olanak tanıyan çok yönlü bir araç setidir.

FastAPI ve SQLAlchemy mükemmel bir eşleşmedir. Güçlü ve benzersiz uygulamaların yaratılmasına olanak tanıyan güvenilir, performanslı ve modern teknolojilerdir. Bu makale, ORM olarak SQLAlchemy 2.0'ı kullanan bir FastAPI arka uç uygulaması oluşturmayı araştırıyor. İçerik şunları kapsar:

• Mapped ve mapped_column kullanarak modeller oluşturma
• soyut bir modelin tanımlanması
• veritabanı oturumunu yönetme
• ORM'yi kullanma
• tüm modeller için ortak bir depo sınıfı oluşturmak
• test kurulumu hazırlama ve testler ekleme

Daha sonra FastAPI uygulamasını SQLAlchemy ORM ile kolaylıkla birleştirebileceksiniz. Ayrıca iyi yapılandırılmış, sağlam ve performanslı uygulamalar oluşturmaya yönelik en iyi uygulamalara ve kalıplara aşina olacaksınız.

Önkoşullar

Makalede yer alan kod örnekleri, içerik ve iksir nesneleri oluşturmaya ve okumaya yönelik temel bir API olan simyacı projesinden gelmektedir. Makalenin ana odağı FastAPI ve SQLAlchemy kombinasyonunu keşfetmektir. Aşağıdakiler gibi diğer konuları kapsamaz:

• Docker kurulumunu yapılandırma
• uvicorn sunucusunu başlatıyorum
• Linting'i ayarlama

Bu konulara ilginiz varsa kod tabanını inceleyerek bunları kendi başınıza keşfedebilirsiniz. Alchemist projesinin kod deposuna erişmek için lütfen buradaki bağlantıyı takip edin. Ayrıca projenin dosya yapısını aşağıda bulabilirsiniz:

 alchemist ├─ alchemist │ ├─ api │ │ ├─ v1 │ │ │ ├─ __init__.py │ │ │ └─ routes.py │ │ ├─ v2 │ │ │ ├─ __init__.py │ │ │ ├─ dependencies.py │ │ │ └─ routes.py │ │ ├─ __init__.py │ │ └─ models.py │ ├─ database │ │ ├─ __init__.py │ │ ├─ models.py │ │ ├─ repository.py │ │ └─ session.py │ ├─ __init__.py │ ├─ app.py │ └─ config.py ├─ requirements │ ├─ base.txt │ └─ dev.txt ├─ scripts │ ├─ create_test_db.sh │ ├─ migrate.py │ └─ run.sh ├─ tests │ ├─ conftest.py │ └─ test_api.py ├─ .env ├─ .gitignore ├─ .pre-commit-config.yaml ├─ Dockerfile ├─ Makefile ├─ README.md ├─ docker-compose.yaml ├─ example.env └─ pyproject.toml

Ağaç büyük görünse de içeriğin bir kısmı bu makalenin ana fikriyle alakalı değil. Ayrıca kod belirli alanlarda gerekenden daha basit görünebilir. Örneğin projede şunlar eksik:

• Dockerfile'da üretim aşaması
• geçişler için imbik kurulumu
• testler için alt dizinler

Bu, karmaşıklığı azaltmak ve gereksiz yüklerden kaçınmak için kasıtlı olarak yapıldı. Ancak üretime daha hazır bir projeyle uğraşıyorsanız bu faktörleri akılda tutmak önemlidir.

API Gereksinimleri

Bir uygulama geliştirmeye başladığınızda uygulamanızın kullanacağı modelleri dikkate almak çok önemlidir. Bu modeller, uygulamanızın birlikte çalışacağı ve API'de gösterileceği nesneleri ve varlıkları temsil edecektir. Simyacı uygulamasında iki varlık vardır: malzemeler ve iksirler. API, bu varlıkların oluşturulmasına ve alınmasına izin vermelidir. alchemist/api/models.py dosyası API'de kullanılacak modelleri içerir:

 import uuid from pydantic import BaseModel, Field class Ingredient(BaseModel): """Ingredient model.""" pk: uuid.UUID name: str class Config: orm_mode = True class IngredientPayload(BaseModel): """Ingredient payload model.""" name: str = Field(min_length=1, max_length=127) class Potion(BaseModel): """Potion model.""" pk: uuid.UUID name: str ingredients: list[Ingredient] class Config: orm_mode = True class PotionPayload(BaseModel): """Potion payload model.""" name: str = Field(min_length=1, max_length=127) ingredients: list[uuid.UUID] = Field(min_items=1)

API, Ingredient ve Potion modellerini döndürecek. Yapılandırmada orm_mode True olarak ayarlanması gelecekte SQLAlchemy nesneleriyle çalışmayı kolaylaştıracaktır. Yeni nesneler oluşturmak için Payload modelleri kullanılacaktır.

Pydantic kullanımı, sınıfların rolleri ve işlevleri açısından daha ayrıntılı ve net olmasını sağlar. Şimdi veritabanı modellerini oluşturmanın zamanı geldi.

Modelleri Bildirme

Model aslında bir şeyin temsilidir . API'ler bağlamında modeller, arka ucun istek gövdesinden ne beklediğini ve yanıt verilerinde ne döndüreceğini temsil eder. Veritabanı modelleri ise daha karmaşıktır ve veritabanında depolanan veri yapılarını ve bunlar arasındaki ilişki türlerini temsil eder.

alchemist/database/models.py dosyası içerik ve iksir nesneleri için modeller içerir:

 import uuid from sqlalchemy import Column, ForeignKey, Table, orm from sqlalchemy.dialects.postgresql import UUID class Base(orm.DeclarativeBase): """Base database model.""" pk: orm.Mapped[uuid.UUID] = orm.mapped_column( primary_key=True, default=uuid.uuid4, ) potion_ingredient_association = Table( "potion_ingredient", Base.metadata, Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")), Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")), ) class Ingredient(Base): """Ingredient database model.""" __tablename__ = "ingredient" name: orm.Mapped[str] class Potion(Base): """Potion database model.""" __tablename__ = "potion" name: orm.Mapped[str] ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship( secondary=potion_ingredient_association, backref="potions", lazy="selectin", )

SQLAlchemy'deki her model DeclarativeBase sınıfıyla başlar. Ondan miras almak, Python türü denetleyicilerle uyumlu veritabanı modelleri oluşturmaya olanak tanır.

Tüm modellerde gerekli alanları içeren soyut bir model (bu durumda Base sınıf) oluşturmak da iyi bir uygulamadır. Bu alanlar, her nesnenin benzersiz tanımlayıcısı olan birincil anahtarı içerir. Soyut model genellikle bir nesne oluşturulduğunda veya güncellendiğinde otomatik olarak ayarlanan nesnenin oluşturulma ve güncellenme tarihlerini de saklar. Ancak Base model basit tutulacaktır.

Ingredient modeline geçildiğinde, __tablename__ niteliği veritabanı tablosunun adını belirtirken, name alanı yeni SQLAlchemy sözdizimini kullanarak model alanlarının tür açıklamalarıyla bildirilmesine olanak tanır. Bu kısa ve modern yaklaşım, name alanını bir dize olarak tanıdığından, tür denetleyicileri ve IDE'ler için hem güçlü hem de avantajlıdır.

Potion modelinde işler daha da karmaşıklaşıyor. Aynı zamanda __tablename__ ve name niteliklerini de içerir, ancak bunun da ötesinde, içeriklerle olan ilişkiyi de saklar. Mapped[list["Ingredient"]] kullanımı, iksirin birden fazla bileşen içerebileceğini ve bu durumda ilişkinin çoktan çoğa (M2M) olduğunu gösterir. Bu, tek bir bileşenin birden fazla iksire atanabileceği anlamına gelir.

M2M, genellikle iki varlık arasındaki bağlantıları saklayan bir ilişki tablosunun oluşturulmasını içeren ek yapılandırma gerektirir. Bu durumda, potion_ingredient_association nesnesi yalnızca içerik maddesinin ve iksirin tanımlayıcılarını saklar ancak aynı zamanda iksir için gereken belirli bir malzemenin miktarı gibi ekstra nitelikleri de içerebilir.

relationship işlevi, iksir ve içindekiler arasındaki ilişkiyi yapılandırır. lazy argümanı ilgili öğelerin nasıl yüklenmesi gerektiğini belirtir. Başka bir deyişle: Bir iksir alırken SQLAlchemy ilgili malzemelerle ne yapmalıdır? Bunu selectin olarak ayarlamak, bileşenlerin iksirle yükleneceği anlamına gelir ve kodda ek sorgu ihtiyacını ortadan kaldırır.

Bir ORM ile çalışırken iyi tasarlanmış modeller oluşturmak çok önemlidir. Bu yapıldıktan sonraki adım, veritabanıyla bağlantının kurulmasıdır.

Oturum İşleyicisi

Bir veritabanıyla çalışırken, özellikle SQLAlchemy kullanırken aşağıdaki kavramları anlamak önemlidir:

• lehçe
• motor
• bağlantı
• bağlantı havuzu
• oturum

Tüm bu terimlerden en önemlisi motordur . SQLAlchemy belgelerine göre, motor nesnesi, veritabanı bağlantısını ve davranışını kolaylaştırmak için Pool ve Dialect bağlamaktan sorumludur. Daha basit bir ifadeyle, motor nesnesi veritabanı bağlantısının kaynağıdır; bağlantı ise SQL ifadelerinin yürütülmesi, işlemlerin yönetilmesi ve veritabanından sonuçların alınması gibi üst düzey işlevler sağlar.

Oturum , ilgili işlemleri tek bir işlem içinde gruplandıran bir iş birimidir. Temel veritabanı bağlantıları üzerinde bir soyutlamadır ve bağlantıları ve işlem davranışını verimli bir şekilde yönetir.

Dialect , belirli bir veritabanı arka ucu için destek sağlayan bir bileşendir. İletişimin ayrıntılarını yöneterek SQLAlchemy ile veritabanı arasında aracı görevi görür. Alchemist projesi veritabanı olarak Postgres'i kullanıyor, dolayısıyla lehçenin bu özel veritabanı türüyle uyumlu olması gerekiyor.

Son soru işareti ise bağlantı havuzudur . SQLAlchemy bağlamında bağlantı havuzu, veritabanı bağlantıları koleksiyonunu yöneten bir mekanizmadır. Her istek için yeni bağlantılar oluşturmak yerine mevcut bağlantıları yeniden kullanarak veritabanı işlemlerinin performansını ve verimliliğini artırmak için tasarlanmıştır. Bağlantı havuzu, bağlantıları yeniden kullanarak yeni bağlantılar kurma ve bunları bozma yükünü azaltır ve bu da performansın artmasını sağlar.

Bu bilgiyi kapsadıktan sonra, artık veritabanına bağlanmak için bağımlılık olarak kullanılacak bir işlevi içeren alchemist/database/session.py dosyasına göz atabilirsiniz:

 from collections.abc import AsyncGenerator from sqlalchemy import exc from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.config import settings async def get_db_session() -> AsyncGenerator[AsyncSession, None]: engine = create_async_engine(settings.DATABASE_URL) factory = async_sessionmaker(engine) async with factory() as session: try: yield session await session.commit() except exc.SQLAlchemyError as error: await session.rollback() raise

Dikkat edilmesi gereken ilk önemli detay get_db_session fonksiyonunun bir üreteç fonksiyonu olmasıdır. Bunun nedeni FastAPI bağımlılık sisteminin oluşturucuları desteklemesidir. Sonuç olarak, bu işlev hem başarılı hem de başarısız senaryoları işleyebilir.

get_db_session fonksiyonunun ilk iki satırı bir veritabanı motoru ve bir oturum oluşturur. Ancak oturum nesnesi aynı zamanda içerik yöneticisi olarak da kullanılabilir. Bu size olası istisnalar ve başarılı sonuçlar üzerinde daha fazla kontrol sağlar.

Her ne kadar SQLAlchemy bağlantıların kapatılmasını yönetse de, bağlantı tamamlandıktan sonra bağlantının nasıl işleneceğini açıkça bildirmek iyi bir uygulamadır. get_db_session işlevinde, her şey yolunda giderse oturum yürütülür ve bir istisna ortaya çıkarsa oturum geri alınır.

Bu kodun asyncio uzantısı etrafında oluşturulduğunu unutmamak önemlidir. SQLAlchemy'nin bu özelliği, uygulamanın veritabanıyla eşzamansız olarak etkileşime girmesine olanak tanır. Bu, veritabanına yapılan isteklerin diğer API isteklerini engellemeyeceği ve uygulamayı daha verimli hale getireceği anlamına gelir.

Modeller ve bağlantı kurulduktan sonraki adım, modellerin veritabanına eklenmesini sağlamaktır.

Hızlı Geçişler

SQLAlchemy modelleri bir veritabanının yapılarını temsil eder. Ancak bunları basitçe oluşturmak veritabanında anında değişikliklere yol açmaz. Değişiklik yapmak için önce bunları uygulamanız gerekir. Bu genellikle her modeli izleyen ve veritabanını buna göre güncelleyen alembic gibi bir geçiş kitaplığı kullanılarak yapılır.

Bu senaryoda modellerde başka değişiklik planlanmadığından temel bir geçiş komut dosyası yeterli olacaktır. Aşağıda scripts/migrate.py dosyasından bir örnek kod bulunmaktadır.

 import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine from alchemist.config import settings from alchemist.database.models import Base logger = logging.getLogger() async def migrate_tables() -> None: logger.info("Starting to migrate") engine = create_async_engine(settings.DATABASE_URL) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Done migrating") if __name__ == "__main__": asyncio.run(migrate_tables())

Basitçe açıklamak gerekirse, migrate_tables işlevi modellerin yapısını okur ve bunu SQLAlchemy motorunu kullanarak veritabanında yeniden oluşturur. Bu betiği çalıştırmak için python scripts/migrate.py komutunu kullanın.

Modeller artık hem kodda hem de veritabanında mevcuttur ve get_db_session veritabanıyla etkileşimi kolaylaştırabilir. Artık API mantığı üzerinde çalışmaya başlayabilirsiniz.

ORM ile API

Daha önce de belirtildiği gibi, malzemeler ve iksirlere yönelik API'nin üç işlemi desteklemesi amaçlanmaktadır:

• nesneler oluşturma
• nesneleri listeleme
• Nesneleri kimliğe göre alma

Önceki hazırlıklar sayesinde tüm bu özellikler, ORM olarak SQLAlchemy ve web çerçevesi olarak FastAPI ile halihazırda uygulanabilmektedir. Başlamak için alchemist/api/v1/routes.py dosyasında bulunan içerik API'sini inceleyin:

 import uuid from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.api import models from alchemist.database import models as db_models from alchemist.database.session import get_db_session router = APIRouter(prefix="/v1", tags=["v1"]) @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = db_models.Ingredient(**data.dict()) session.add(ingredient) await session.commit() await session.refresh(ingredient) return models.Ingredient.from_orm(ingredient) @router.get("/ingredients", status_code=status.HTTP_200_OK) async def get_ingredients( session: AsyncSession = Depends(get_db_session), ) -> list[models.Ingredient]: ingredients = await session.scalars(select(db_models.Ingredient)) return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients] @router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK) async def get_ingredient( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = await session.get(db_models.Ingredient, pk) if ingredient is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient does not exist", ) return models.Ingredient.from_orm(ingredient)

/ingredients API'sinin altında üç yol mevcuttur. POST uç noktası, daha önce oluşturulmuş bir modelden ve bir veritabanı oturumundan bir nesne olarak bir içerik yükünü alır. get_db_session oluşturucu işlevi oturumu başlatır ve veritabanı etkileşimlerini etkinleştirir.

Gerçek işlev gövdesinde gerçekleşen beş adım vardır:

1. Gelen veriden bir içerik nesnesi oluşturulur.
2. Oturum nesnesinin add yöntemi, içerik nesnesini oturum izleme sistemine ekler ve onu veritabanına eklenmek üzere beklemede olarak işaretler.
3. Oturum tamamlandı.
4. İçerik nesnesi, özniteliklerinin veritabanı durumuyla eşleştiğinden emin olmak için yenilenir.
5. Veritabanı bileşeni örneği, from_orm yöntemi kullanılarak API modeli örneğine dönüştürülür.

Hızlı bir test için çalışan uygulamaya karşı basit bir kıvrılma yürütülebilir:

 curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}'

Yanıtta, veritabanından gelen bir kimliğe sahip bir içerik nesnesi bulunmalıdır:

 { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" }

SQLAlchemy'nin çoklu soyutlama katmanları basit bir API için gereksiz görünse de, ORM ayrıntılarını ayrı tutar ve SQLAlchemy'nin verimliliğine ve ölçeklenebilirliğine katkıda bulunur. Asyncio ile birleştirildiğinde ORM özellikleri API'de olağanüstü iyi performans gösterir.

Geriye kalan iki uç nokta daha az karmaşıktır ve benzerlikleri paylaşmaktadır. Daha derin bir açıklamayı hak eden kısımlardan biri, get_ingredients işlevi içinde scalars yönteminin kullanılmasıdır. SQLAlchemy kullanarak veritabanını sorgularken, execute yöntemi genellikle argüman olarak bir sorguyla birlikte kullanılır. execute yöntemi satır benzeri demetler döndürürken, scalars ORM varlıklarını doğrudan döndürerek uç noktayı daha temiz hale getirir.

Şimdi aynı dosyadaki iksir API'sini düşünün:

 @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: data_dict = data.dict() ingredients = await session.scalars( select(db_models.Ingredient).where( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) ) potion = db_models.Potion(**data_dict, ingredients=list(ingredients)) session.add(potion) await session.commit() await session.refresh(potion) return models.Potion.from_orm(potion) @router.get("/potions", status_code=status.HTTP_200_OK) async def get_potions( session: AsyncSession = Depends(get_db_session), ) -> list[models.Potion]: potions = await session.scalars(select(db_models.Potion)) return [models.Potion.from_orm(potion) for potion in potions] @router.get("/potions/{pk}", status_code=status.HTTP_200_OK) async def get_potion( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: potion = await session.get(db_models.Potion, pk) if potion is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Potion does not exist", ) return models.Potion.from_orm(potion)

İksirler için GET uç noktaları, malzemeler için olanlarla aynıdır. Ancak POST işlevi ek kod gerektirir. Bunun nedeni, iksir oluşturmanın en az bir içerik kimliğinin eklenmesini gerektirmesidir; bu da malzemelerin getirilip yeni oluşturulan iksire bağlanması gerektiği anlamına gelir. Bunu başarmak için yine scalars yöntem kullanılır, ancak bu kez getirilen bileşenlerin kimliklerini belirten bir sorgu kullanılır. İksir oluşturma sürecinin geri kalan kısmı malzemelerle aynıdır.

Uç noktayı test etmek için yine bir curl komutu çalıştırılabilir.

 curl -X 'POST' \ 'http://localhost:8000/api/v1/potions' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}'

Aşağıdaki yanıtla sonuçlanır:

 { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] }

İlişkide belirtilen lazy="selectin" argümanı sayesinde, her bir bileşenin iksir içinde tam bir nesne olarak temsil edildiğine dikkat etmek önemlidir.

API'ler işlevseldir ancak kodla ilgili önemli bir sorun vardır. SQLAlchemy size veritabanıyla istediğiniz gibi etkileşimde bulunma özgürlüğü verirken, Django'nun Model.objects benzer herhangi bir üst düzey "yönetici" yardımcı programı sunmaz. Sonuç olarak, onu kendiniz oluşturmanız gerekecek; bu da esasen içerik ve iksir API'lerinde kullanılan mantıktır. Ancak bu mantığı ayrı bir alana çıkarmadan doğrudan uç noktalarda tutarsanız, çok sayıda kopya kodla karşılaşırsınız. Ayrıca sorgularda veya modellerde değişiklik yapmanın yönetilmesi giderek zorlaşacaktır.

Gelecek bölümde depo modeli tanıtılıyor: ORM kodunu çıkarmak için zarif bir çözüm.

Depo

Depo modeli, veritabanıyla çalışmanın ayrıntılarını soyutlamaya olanak tanır. Simyacı örneğinde olduğu gibi SQLAlchemy kullanılması durumunda, depo sınıfı birden fazla modeli yönetmekten ve veritabanı oturumuyla etkileşim kurmaktan sorumlu olacaktır.

alchemist/database/repository.py dosyasındaki aşağıdaki koda bir göz atın:

 import uuid from typing import Generic, TypeVar from sqlalchemy import BinaryExpression, select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models Model = TypeVar("Model", bound=models.Base) class DatabaseRepository(Generic[Model]): """Repository for performing database queries.""" def __init__(self, model: type[Model], session: AsyncSession) -> None: self.model = model self.session = session async def create(self, data: dict) -> Model: instance = self.model(**data) self.session.add(instance) await self.session.commit() await self.session.refresh(instance) return instance async def get(self, pk: uuid.UUID) -> Model | None: return await self.session.get(self.model, pk) async def filter( self, *expressions: BinaryExpression, ) -> list[Model]: query = select(self.model) if expressions: query = query.where(*expressions) return list(await self.session.scalars(query))

DatabaseRepository sınıfı, daha önce uç noktalara dahil edilen tüm mantığı tutar. Aradaki fark, belirli model sınıfının __init__ yöntemiyle aktarılmasına izin vermesi ve kodu her uç noktada çoğaltmak yerine tüm modeller için yeniden kullanılmasını mümkün kılmasıdır.

Ayrıca DatabaseRepository , Python jeneriklerini kullanır ve Model jenerik türü soyut veritabanı modeline bağlanır. Bu, depo sınıfının statik tür kontrolünden daha fazla yararlanmasına olanak tanır. Belirli bir modelle kullanıldığında depo yöntemlerinin dönüş türleri bu özel modeli yansıtacaktır.

Havuzun veritabanı oturumunu kullanması gerektiğinden, get_db_session bağımlılığıyla birlikte başlatılması gerekir. alchemist/api/v2/dependencies.py dosyasındaki yeni bağımlılığı göz önünde bulundurun:

 from collections.abc import Callable from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models, repository, session def get_repository(  model: type[models.Base], ) -> Callable[[AsyncSession], repository.DatabaseRepository]:  def func(session: AsyncSession = Depends(session.get_db_session)):    return repository.DatabaseRepository(model, session)  return func

Basitçe söylemek gerekirse get_repository işlevi bir bağımlılık fabrikasıdır. Öncelikle repository'yi kullanacağınız veritabanı modelini alır. Daha sonra veritabanı oturumunu almak ve depo nesnesini başlatmak için kullanılacak bağımlılığı döndürür. Daha iyi anlamak için alchemist/api/v2/routes.py dosyasındaki yeni API'ye göz atın. Yalnızca POST uç noktalarını gösterir, ancak kodun nasıl geliştirildiğine dair size daha net bir fikir vermek için yeterli olacaktır:

 from typing import Annotated from fastapi import APIRouter, Depends, status from alchemist.api import models from alchemist.api.v2.dependencies import get_repository from alchemist.database import models as db_models from alchemist.database.repository import DatabaseRepository router = APIRouter(prefix="/v2", tags=["v2"]) IngredientRepository = Annotated[  DatabaseRepository[db_models.Ingredient],  Depends(get_repository(db_models.Ingredient)), ] PotionRepository = Annotated[  DatabaseRepository[db_models.Potion],  Depends(get_repository(db_models.Potion)), ] @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, repository: IngredientRepository, ) -> models.Ingredient: ingredient = await repository.create(data.dict()) return models.Ingredient.from_orm(ingredient) @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, ingredient_repository: IngredientRepository, potion_repository: PotionRepository, ) -> models.Potion: data_dict = data.dict() ingredients = await ingredient_repository.filter( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) potion = await potion_repository.create({**data_dict, "ingredients": ingredients}) return models.Potion.from_orm(potion)

Dikkat edilmesi gereken ilk önemli özellik, FastAPI bağımlılıklarıyla çalışmanın yeni bir yolu olan Annotated kullanılmasıdır. Bağımlılığın dönüş türünü DatabaseRepository[db_models.Ingredient] olarak belirterek ve kullanımını Depends(get_repository(db_models.Ingredient)) ile bildirerek, uç noktadaki basit tür ek açıklamalarını kullanmaya son verebilirsiniz: repository: IngredientRepository .

Depo sayesinde uç noktaların ORM ile ilgili tüm yükü depolaması gerekmez. Daha karmaşık iksir durumunda bile tek yapmanız gereken iki depoyu aynı anda kullanmaktır.

İki havuzun başlatılmasının oturumu iki kez başlatıp başlatmayacağını merak edebilirsiniz. Cevap hayır. FastAPI bağımlılık sistemi aynı bağımlılık çağrılarını tek bir istekte önbelleğe alır. Bu, oturum başlatmanın önbelleğe alındığı ve her iki havuzun da tam olarak aynı oturum nesnesini kullandığı anlamına gelir. SQLAlchemy ve FastAPI kombinasyonunun bir başka harika özelliği.

API tamamen işlevseldir ve yeniden kullanılabilir, yüksek performanslı bir veri erişim katmanına sahiptir. Bir sonraki adım, bazı uçtan uca testler yazarak gereksinimlerin karşılandığından emin olmaktır.

Test yapmak

Testler yazılım geliştirmede çok önemli bir rol oynar. Projeler birim, entegrasyon ve uçtan uca (E2E) testleri içerebilir. Yüksek sayıda anlamlı birim testine sahip olmak genellikle en iyisi olsa da, tüm iş akışının doğru şekilde çalıştığından emin olmak için en az birkaç E2E testi yazmak da iyidir.

Alchemist uygulamasına yönelik bazı E2E testleri oluşturmak için iki ek kütüphane gereklidir:

• testleri gerçekten oluşturmak ve çalıştırmak için pytest
• Testlerin içinde eşzamansız istekler yapmak için httpx

Bunlar kurulduktan sonra bir sonraki adım ayrı bir test veritabanına sahip olmaktır. Varsayılan veritabanınızın kirlenmesini veya düşmesini istemezsiniz. Alchemist bir Docker kurulumu içerdiğinden, ikinci bir veritabanı oluşturmak için yalnızca basit bir komut dosyasına ihtiyaç vardır. scripts/create_test_db.sh dosyasındaki koda bir göz atın:

 #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test"

Komut dosyasının yürütülebilmesi için Postgres konteynerine birim olarak eklenmesi gerekir. Bu, docker-compose.yaml dosyasının volumes bölümüne dahil edilerek gerçekleştirilebilir.

Hazırlığın son adımı, tests/conftest.py dosyasında pytest fikstürleri oluşturmaktır:

 from collections.abc import AsyncGenerator import pytest import pytest_asyncio from fastapi import FastAPI from httpx import AsyncClient from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.app import app from alchemist.config import settings from alchemist.database.models import Base from alchemist.database.session import get_db_session @pytest_asyncio.fixture() async def db_session() -> AsyncGenerator[AsyncSession, None]: """Start a test database session.""" db_name = settings.DATABASE_URL.split("/")[-1] db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test") engine = create_async_engine(db_url) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) session = async_sessionmaker(engine)() yield session await session.close() @pytest.fixture() def test_app(db_session: AsyncSession) -> FastAPI: """Create a test app with overridden dependencies.""" app.dependency_overrides[get_db_session] = lambda: db_session return app @pytest_asyncio.fixture() async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]: """Create an http client.""" async with AsyncClient(app=test_app, base_url="http://test") as client: yield client

Testlerde değiştirilmesi gereken şeylerden biri uygulamanın veritabanıyla nasıl etkileşime girdiğidir. Bu, yalnızca veritabanı URL'sini değiştirmeyi değil aynı zamanda boş bir veritabanıyla başlayarak her testin izole edilmesini de içerir.

db_session fikstürü bu hedeflerin her ikisini de gerçekleştirir. Vücudu aşağıdaki adımları atar:

1. Değiştirilmiş bir veritabanı URL'sine sahip bir motor oluşturun.
2. Testin temiz bir veritabanına sahip olduğundan emin olmak için mevcut tüm tabloları silin.
3. Veritabanındaki tüm tabloları oluşturun (geçiş komut dosyasındaki kodun aynısı).
4. Bir oturum nesnesi oluşturun ve sağlayın.
5. Test tamamlandığında oturumu manuel olarak kapatın.

Her ne kadar son adım bağlam yöneticisi olarak da uygulanabilse de, bu durumda manuel kapatma gayet iyi çalışıyor.

Geriye kalan iki fikstür oldukça açıklayıcı olmalıdır:

• test_app alchemist/app.py dosyasındaki FastAPI örneğidir ve get_db_session bağımlılığı db_session fikstürüyle değiştirilmiştir.
• client , test_app karşı API istekleri yapacak httpx AsyncClient

Tüm bunlar ayarlandıktan sonra nihayet gerçek testler yazılabilir. Kısa ve öz olması açısından, tests/test_api.py dosyasından alınan aşağıdaki örnekte yalnızca bir içerik oluşturmaya yönelik bir test gösterilmektedir:

 from fastapi import status class TestIngredientsAPI: """Test cases for the ingredients API.""" async def test_create_ingredient(self, client): response = await client.post("/api/v2/ingredients", json={"name": "Carrot"}) assert response.status_code == status.HTTP_201_CREATED pk = response.json().get("pk") assert pk is not None response = await client.get("/api/v2/ingredients") assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 1 assert response.json()[0]["pk"] == pk

Test, FastAPI örneğine geçersiz kılınan bağımlılıkla istekte bulunan, bir fikstürde oluşturulan bir istemci nesnesini kullanır. Sonuç olarak test, test tamamlandıktan sonra temizlenecek ayrı bir veritabanıyla etkileşime girebilir. Her iki API için de kalan test paketinin yapısı hemen hemen aynıdır.

Özet

FastAPI ve SQLAlchemy, modern ve güçlü arka uç uygulamaları oluşturmaya yönelik mükemmel teknolojilerdir. Sundukları özgürlük, basitlik ve esneklik, onları Python tabanlı projeler için en iyi seçeneklerden biri haline getiriyor. Geliştiriciler en iyi uygulamaları ve kalıpları takip ederse veritabanı işlemlerini ve API mantığını kolaylıkla yürüten performanslı, sağlam ve iyi yapılandırılmış uygulamalar oluşturabilirler. Bu makale, bu muhteşem kombinasyonun nasıl kurulacağı ve sürdürüleceği konusunda size iyi bir anlayış sağlamayı amaçladı.

Kaynaklar

Alchemist projesinin kaynak kodunu burada bulabilirsiniz: link .