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, 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. FastAPI 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 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. SQLAlchemy'nin 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: ve kullanarak modeller oluşturma Mapped mapped_column 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 projesinden gelmektedir. Makalenin ana odağı FastAPI ve SQLAlchemy kombinasyonunu keşfetmektir. Aşağıdakiler gibi diğer konuları kapsamaz: simyacı 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 bağlantıyı takip edin. Ayrıca projenin dosya yapısını aşağıda bulabilirsiniz: buradaki 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 ve 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. dosyası API'de kullanılacak modelleri içerir: nesneleri varlıkları alchemist/api/models.py 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, ve modellerini döndürecek. Yapılandırmada olarak ayarlanması gelecekte SQLAlchemy nesneleriyle çalışmayı kolaylaştıracaktır. Yeni nesneler oluşturmak için modelleri kullanılacaktır. Ingredient Potion orm_mode True Payload 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 . 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 ve bunlar arasındaki türlerini temsil eder. temsilidir veri yapılarını ilişki dosyası içerik ve iksir nesneleri için modeller içerir: alchemist/database/models.py 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 sınıfıyla başlar. Ondan miras almak, Python türü denetleyicilerle uyumlu veritabanı modelleri oluşturmaya olanak tanır. DeclarativeBase Tüm modellerde gerekli alanları içeren bir model (bu durumda 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 model basit tutulacaktır. soyut Base Base modeline geçildiğinde, niteliği veritabanı tablosunun adını belirtirken, 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, alanını bir dize olarak tanıdığından, tür denetleyicileri ve IDE'ler için hem güçlü hem de avantajlıdır. Ingredient __tablename__ name name modelinde işler daha da karmaşıklaşıyor. Aynı zamanda ve niteliklerini de içerir, ancak bunun da ötesinde, içeriklerle olan ilişkiyi de saklar. 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. Potion __tablename__ name Mapped[list["Ingredient"]] 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, 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. potion_ingredient_association işlevi, iksir ve içindekiler arasındaki ilişkiyi yapılandırır. 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 olarak ayarlamak, bileşenlerin iksirle yükleneceği anlamına gelir ve kodda ek sorgu ihtiyacını ortadan kaldırır. relationship lazy selectin 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 . SQLAlchemy belgelerine göre, motor nesnesi, veritabanı bağlantısını ve davranışını kolaylaştırmak için ve bağlamaktan sorumludur. Daha basit bir ifadeyle, motor nesnesi veritabanı bağlantısının kaynağıdır; 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. motordur Pool Dialect bağlantı , 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. Oturum , 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. Dialect Son soru işareti ise . 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. bağlantı havuzudur 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 dosyasına göz atabilirsiniz: alchemist/database/session.py 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 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. get_db_session 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. işlevinde, her şey yolunda giderse oturum yürütülür ve bir istisna ortaya çıkarsa oturum geri alınır. get_db_session 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ı 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. uygulamanız Bu senaryoda modellerde başka değişiklik planlanmadığından temel bir geçiş komut dosyası yeterli olacaktır. Aşağıda dosyasından bir örnek kod bulunmaktadır. scripts/migrate.py 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, 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 komutunu kullanın. migrate_tables python scripts/migrate.py Modeller artık hem kodda hem de veritabanında mevcuttur ve veritabanıyla etkileşimi kolaylaştırabilir. Artık API mantığı üzerinde çalışmaya başlayabilirsiniz. get_db_session 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 dosyasında bulunan içerik API'sini inceleyin: alchemist/api/v1/routes.py 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) 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. oluşturucu işlevi oturumu başlatır ve veritabanı etkileşimlerini etkinleştirir. /ingredients get_db_session Gerçek işlev gövdesinde gerçekleşen beş adım vardır: Gelen veriden bir içerik nesnesi oluşturulur. Oturum nesnesinin yöntemi, içerik nesnesini oturum izleme sistemine ekler ve onu veritabanına eklenmek üzere beklemede olarak işaretler. add Oturum tamamlandı. İçerik nesnesi, özniteliklerinin veritabanı durumuyla eşleştiğinden emin olmak için yenilenir. Veritabanı bileşeni örneği, yöntemi kullanılarak API modeli örneğine dönüştürülür. from_orm 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, işlevi içinde yönteminin kullanılmasıdır. SQLAlchemy kullanarak veritabanını sorgularken, yöntemi genellikle argüman olarak bir sorguyla birlikte kullanılır. yöntemi satır benzeri demetler döndürürken, ORM varlıklarını doğrudan döndürerek uç noktayı daha temiz hale getirir. get_ingredients scalars execute execute scalars Ş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 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. scalars 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 argümanı sayesinde, her bir bileşenin iksir içinde tam bir nesne olarak temsil edildiğine dikkat etmek önemlidir. lazy="selectin" 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 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. Model.objects Gelecek bölümde depo modeli tanıtılıyor: ORM kodunu çıkarmak için zarif bir çözüm. 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. Depo dosyasındaki aşağıdaki koda bir göz atın: alchemist/database/repository.py 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)) sınıfı, daha önce uç noktalara dahil edilen tüm mantığı tutar. Aradaki fark, belirli model sınıfının 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. DatabaseRepository __init__ Ayrıca , Python jeneriklerini kullanır ve 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. DatabaseRepository Model Havuzun veritabanı oturumunu kullanması gerektiğinden, bağımlılığıyla birlikte başlatılması gerekir. dosyasındaki yeni bağımlılığı göz önünde bulundurun: get_db_session alchemist/api/v2/dependencies.py 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 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 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: get_repository alchemist/api/v2/routes.py 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 kullanılmasıdır. Bağımlılığın dönüş türünü olarak belirterek ve kullanımını ile bildirerek, uç noktadaki basit tür ek açıklamalarını kullanmaya son verebilirsiniz: . Annotated DatabaseRepository[db_models.Ingredient] Depends(get_repository(db_models.Ingredient)) 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. dosyasındaki koda bir göz atın: scripts/create_test_db.sh #!/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, dosyasının bölümüne dahil edilerek gerçekleştirilebilir. docker-compose.yaml volumes Hazırlığın son adımı, dosyasında pytest fikstürleri oluşturmaktır: tests/conftest.py 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. fikstürü bu hedeflerin her ikisini de gerçekleştirir. Vücudu aşağıdaki adımları atar: db_session Değiştirilmiş bir veritabanı URL'sine sahip bir motor oluşturun. Testin temiz bir veritabanına sahip olduğundan emin olmak için mevcut tüm tabloları silin. Veritabanındaki tüm tabloları oluşturun (geçiş komut dosyasındaki kodun aynısı). Bir oturum nesnesi oluşturun ve sağlayın. 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: dosyasındaki FastAPI örneğidir ve bağımlılığı fikstürüyle değiştirilmiştir. test_app alchemist/app.py get_db_session db_session , karşı API istekleri yapacak httpx client test_app AsyncClient Tüm bunlar ayarlandıktan sonra nihayet gerçek testler yazılabilir. Kısa ve öz olması açısından, dosyasından alınan aşağıdaki örnekte yalnızca bir içerik oluşturmaya yönelik bir test gösterilmektedir: tests/test_api.py 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 da yayınlandı. Burada

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

Çok uzun; Okumak

@tobi

Credibility

İLGİLİ ÖYKÜLER