फास्टएपीआई के साथ SQLAlchemy 2.0 का उपयोग करने के लिए पैटर्न और अभ्यास

जबकि Django और फ्लास्क कई पायथन इंजीनियरों के लिए पहली पसंद बने हुए हैं, फास्टएपीआई को पहले से ही एक निर्विवाद रूप से विश्वसनीय पसंद के रूप में मान्यता दी गई है। यह एक अत्यधिक लचीला, अच्छी तरह से अनुकूलित, संरचित ढांचा है जो डेवलपर को बैकएंड एप्लिकेशन बनाने की अनंत संभावनाएं देता है।

डेटाबेस के साथ काम करना अधिकांश बैकएंड अनुप्रयोगों का एक अनिवार्य पहलू है। परिणामस्वरूप, ORM बैकएंड कोड में एक महत्वपूर्ण भूमिका निभाता है। हालाँकि, Django के विपरीत, FastAPI में ORM बिल्ट-इन नहीं है। एक उपयुक्त लाइब्रेरी का चयन करना और उसे कोडबेस में एकीकृत करना पूरी तरह से डेवलपर की जिम्मेदारी है।

पायथन इंजीनियर व्यापक रूप से SQLAlchemy को सबसे लोकप्रिय उपलब्ध ORM मानते हैं। यह एक प्रसिद्ध पुस्तकालय है जिसका उपयोग 2006 से किया जा रहा है और इसे हजारों परियोजनाओं द्वारा अपनाया गया है। 2023 में, इसे संस्करण 2.0 में एक बड़ा अपडेट प्राप्त हुआ। फास्टएपीआई के समान, SQLAlchemy डेवलपर्स को शक्तिशाली सुविधाओं और उपयोगिताओं को एक विशिष्ट तरीके से उपयोग करने के लिए मजबूर किए बिना प्रदान करता है। अनिवार्य रूप से, यह एक बहुमुखी टूलकिट है जो डेवलपर्स को इसे अपनी इच्छानुसार उपयोग करने का अधिकार देता है।

फास्टएपीआई और एसक्यूएलकेमी स्वर्ग में बना एक मेल है। वे दोनों विश्वसनीय, निष्पादन योग्य और आधुनिक प्रौद्योगिकियां हैं, जो शक्तिशाली और अद्वितीय अनुप्रयोगों के निर्माण को सक्षम बनाती हैं। यह आलेख एक फास्टएपीआई बैकएंड एप्लिकेशन बनाने की पड़ताल करता है जो ओआरएम के रूप में SQLAlchemy 2.0 का उपयोग करता है। सामग्री में शामिल हैं:

• Mapped और mapped_column का उपयोग करके मॉडल बनाना
• एक अमूर्त मॉडल को परिभाषित करना
• डेटाबेस सत्र को संभालना
• ओआरएम का उपयोग करना
• सभी मॉडलों के लिए एक सामान्य रिपॉजिटरी क्लास बनाना
• एक परीक्षण सेटअप तैयार करना और परीक्षण जोड़ना

इसके बाद, आप फास्टएपीआई एप्लिकेशन को SQLAlchemy ORM के साथ आसानी से संयोजित कर पाएंगे। इसके अतिरिक्त, आप अच्छी तरह से संरचित, मजबूत और प्रदर्शन करने वाले एप्लिकेशन बनाने के लिए सर्वोत्तम प्रथाओं और पैटर्न से परिचित होंगे।

आवश्यक शर्तें

लेख में शामिल कोड उदाहरण अल्केमिस्ट प्रोजेक्ट से आते हैं, जो घटक और औषधि वस्तुओं को बनाने और पढ़ने के लिए एक बुनियादी एपीआई है। लेख का मुख्य फोकस फास्टएपीआई और एसक्यूएलकेमी के संयोजन का पता लगाना है। इसमें अन्य विषय शामिल नहीं हैं, जैसे:

• डॉकर सेटअप को कॉन्फ़िगर करना
• यूविकॉर्न सर्वर प्रारंभ करना
• लिंटिंग स्थापित करना

यदि आप इन विषयों में रुचि रखते हैं, तो आप कोडबेस की जांच करके स्वयं उनका पता लगा सकते हैं। अल्केमिस्ट प्रोजेक्ट के कोड रिपॉजिटरी तक पहुंचने के लिए, कृपया यहां इस लिंक का अनुसरण करें। इसके अतिरिक्त, आप नीचे प्रोजेक्ट की फ़ाइल संरचना पा सकते हैं:

 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

हालाँकि पेड़ बड़ा लग सकता है, लेकिन कुछ सामग्री इस लेख के मुख्य बिंदु से प्रासंगिक नहीं हैं। इसके अतिरिक्त, कोड कुछ क्षेत्रों में आवश्यकता से अधिक सरल दिखाई दे सकता है। उदाहरण के लिए, परियोजना में कमी है:

• Dockerfile में उत्पादन चरण
• प्रवासन के लिए एलेम्बिक सेटअप
• परीक्षणों के लिए उपनिर्देशिकाएँ

यह जानबूझकर जटिलता को कम करने और अनावश्यक ओवरहेड से बचने के लिए किया गया था। हालाँकि, अधिक उत्पादन-तैयार परियोजना के साथ काम करते समय इन कारकों को ध्यान में रखना महत्वपूर्ण है।

एपीआई आवश्यकताएँ

किसी ऐप को विकसित करना शुरू करते समय, उन मॉडलों पर विचार करना महत्वपूर्ण है जिनका उपयोग आपका ऐप करेगा। ये मॉडल उन वस्तुओं और संस्थाओं का प्रतिनिधित्व करेंगे जिनके साथ आपका ऐप काम करेगा और एपीआई में प्रदर्शित किया जाएगा। अल्केमिस्ट ऐप के मामले में, दो इकाइयाँ हैं: सामग्री और औषधि। एपीआई को इन संस्थाओं को बनाने और पुनः प्राप्त करने की अनुमति देनी चाहिए। alchemist/api/models.py फ़ाइल में वे मॉडल शामिल हैं जिनका उपयोग API में किया जाएगा:

 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)

एपीआई Ingredient और Potion मॉडल लौटाएगा। कॉन्फ़िगरेशन में orm_mode को True पर सेट करने से भविष्य में SQLAlchemy ऑब्जेक्ट के साथ काम करना आसान हो जाएगा। नई वस्तुएं बनाने के लिए Payload मॉडल का उपयोग किया जाएगा।

पाइडेंटिक का उपयोग कक्षाओं को उनकी भूमिकाओं और कार्यों में अधिक विस्तृत और स्पष्ट बनाता है। अब, डेटाबेस मॉडल बनाने का समय आ गया है।

मॉडल घोषित करना

एक मॉडल अनिवार्य रूप से किसी चीज़ का प्रतिनिधित्व है। एपीआई के संदर्भ में, मॉडल दर्शाते हैं कि बैकएंड अनुरोध निकाय में क्या अपेक्षा करता है और यह प्रतिक्रिया डेटा में क्या लौटाएगा। दूसरी ओर, डेटाबेस मॉडल अधिक जटिल होते हैं और डेटाबेस में संग्रहीत डेटा संरचनाओं और उनके बीच संबंध प्रकारों का प्रतिनिधित्व करते हैं।

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 में प्रत्येक मॉडल DeclarativeBase क्लास से शुरू होता है। इसे इनहेरिट करने से ऐसे डेटाबेस मॉडल बनाने की अनुमति मिलती है जो पायथन प्रकार के चेकर्स के साथ संगत हैं।

एक अमूर्त मॉडल - इस मामले में Base क्लास - बनाना भी एक अच्छा अभ्यास है जिसमें सभी मॉडलों में आवश्यक फ़ील्ड शामिल हैं। इन फ़ील्ड में प्राथमिक कुंजी शामिल है, जो प्रत्येक ऑब्जेक्ट का एक विशिष्ट पहचानकर्ता है। अमूर्त मॉडल अक्सर किसी ऑब्जेक्ट के निर्माण और अद्यतन तिथियों को भी संग्रहीत करता है, जो किसी ऑब्जेक्ट के बनने या अपडेट होने पर स्वचालित रूप से सेट हो जाते हैं। हालाँकि, Base मॉडल को सरल रखा जाएगा।

Ingredient मॉडल पर आगे बढ़ते हुए, __tablename__ विशेषता डेटाबेस तालिका का नाम निर्दिष्ट करती है, जबकि name फ़ील्ड नए SQLAlchemy सिंटैक्स का उपयोग करती है, जिससे मॉडल फ़ील्ड को प्रकार एनोटेशन के साथ घोषित किया जा सकता है। यह संक्षिप्त और आधुनिक दृष्टिकोण टाइप चेकर्स और आईडीई के लिए शक्तिशाली और लाभप्रद दोनों है, क्योंकि यह name फ़ील्ड को एक स्ट्रिंग के रूप में पहचानता है।

Potion मॉडल में चीजें अधिक जटिल हो जाती हैं। इसमें __tablename__ और name विशेषताएँ भी शामिल हैं, लेकिन इसके अलावा, यह अवयवों से संबंध संग्रहीत करता है। Mapped[list["Ingredient"]] का उपयोग इंगित करता है कि औषधि में कई सामग्रियां हो सकती हैं, और इस मामले में, संबंध अनेक-से-अनेक (एम2एम) है। इसका मतलब यह है कि एक ही घटक को कई औषधियों में मिलाया जा सकता है।

एम2एम को अतिरिक्त कॉन्फ़िगरेशन की आवश्यकता होती है, जिसमें आमतौर पर एक एसोसिएशन तालिका का निर्माण शामिल होता है जो दो संस्थाओं के बीच कनेक्शन को संग्रहीत करता है। इस मामले में, potion_ingredient_association ऑब्जेक्ट केवल घटक और पोशन के पहचानकर्ताओं को संग्रहीत करता है, लेकिन इसमें अतिरिक्त गुण भी शामिल हो सकते हैं, जैसे कि पोशन के लिए आवश्यक विशिष्ट घटक की मात्रा।

relationship फ़ंक्शन औषधि और उसके अवयवों के बीच संबंध को कॉन्फ़िगर करता है। lazy तर्क निर्दिष्ट करता है कि संबंधित वस्तुओं को कैसे लोड किया जाना चाहिए। दूसरे शब्दों में: जब आप औषधि ला रहे हों तो SQLAlchemy को संबंधित सामग्रियों के साथ क्या करना चाहिए। इसे selectin पर सेट करने का मतलब है कि अवयवों को पोशन के साथ लोड किया जाएगा, जिससे कोड में अतिरिक्त प्रश्नों की आवश्यकता समाप्त हो जाएगी।

ORM के साथ काम करते समय अच्छी तरह से डिज़ाइन किए गए मॉडल बनाना महत्वपूर्ण है। एक बार यह हो जाने के बाद, अगला चरण डेटाबेस के साथ कनेक्शन स्थापित करना है।

सत्र संचालक

डेटाबेस के साथ काम करते समय, विशेष रूप से SQLAlchemy का उपयोग करते समय, निम्नलिखित अवधारणाओं को समझना आवश्यक है:

• बोली
• इंजन
• कनेक्शन
• कनेक्शन पूल
• सत्र

इन सभी शब्दों में से सबसे महत्वपूर्ण है इंजन । SQLAlchemy दस्तावेज़ के अनुसार, इंजन ऑब्जेक्ट डेटाबेस कनेक्टिविटी और व्यवहार को सुविधाजनक बनाने के लिए Pool और Dialect जोड़ने के लिए ज़िम्मेदार है। सरल शब्दों में, इंजन ऑब्जेक्ट डेटाबेस कनेक्शन का स्रोत है, जबकि कनेक्शन SQL स्टेटमेंट निष्पादित करने, लेनदेन प्रबंधित करने और डेटाबेस से परिणाम पुनर्प्राप्त करने जैसी उच्च-स्तरीय कार्यक्षमता प्रदान करता है।

एक सत्र कार्य की एक इकाई है जो एक ही लेन-देन के भीतर संबंधित कार्यों को समूहित करता है। यह अंतर्निहित डेटाबेस कनेक्शन पर एक अमूर्त है और कुशलतापूर्वक कनेक्शन और लेनदेन व्यवहार का प्रबंधन करता है।

बोली एक घटक है जो एक विशिष्ट डेटाबेस बैकएंड के लिए समर्थन प्रदान करता है। यह SQLAlchemy और डेटाबेस के बीच मध्यस्थ के रूप में कार्य करता है, संचार के विवरण को संभालता है। अल्केमिस्ट प्रोजेक्ट पोस्टग्रेज़ को डेटाबेस के रूप में उपयोग करता है, इसलिए बोली इस विशिष्ट डेटाबेस प्रकार के साथ संगत होनी चाहिए।

अंतिम प्रश्न चिह्न कनेक्शन पूल है। SQLAlchemy के संदर्भ में, एक कनेक्शन पूल एक तंत्र है जो डेटाबेस कनेक्शन के संग्रह का प्रबंधन करता है। इसे प्रत्येक अनुरोध के लिए नए कनेक्शन बनाने के बजाय मौजूदा कनेक्शन का पुन: उपयोग करके डेटाबेस संचालन के प्रदर्शन और दक्षता में सुधार करने के लिए डिज़ाइन किया गया है। कनेक्शन का पुन: उपयोग करके, कनेक्शन पूल नए कनेक्शन स्थापित करने और उन्हें तोड़ने के ओवरहेड को कम कर देता है, जिसके परिणामस्वरूप प्रदर्शन में सुधार होता है।

उस ज्ञान को कवर करने के बाद, अब आप 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

ध्यान देने योग्य पहली महत्वपूर्ण बात यह है कि फ़ंक्शन get_db_session एक जेनरेटर फ़ंक्शन है। ऐसा इसलिए है क्योंकि फास्टएपीआई निर्भरता प्रणाली जनरेटर का समर्थन करती है। परिणामस्वरूप, यह फ़ंक्शन सफल और असफल दोनों परिदृश्यों को संभाल सकता है।

get_db_session फ़ंक्शन की पहली दो पंक्तियाँ एक डेटाबेस इंजन और एक सत्र बनाती हैं। हालाँकि, सत्र ऑब्जेक्ट का उपयोग संदर्भ प्रबंधक के रूप में भी किया जा सकता है। इससे आपको संभावित अपवादों और सफल परिणामों पर अधिक नियंत्रण मिलता है।

हालाँकि SQLAlchemy कनेक्शन को बंद करने का काम संभालती है, लेकिन यह स्पष्ट रूप से घोषित करना एक अच्छा अभ्यास है कि कनेक्शन पूरा होने के बाद उसे कैसे संभालना है। get_db_session फ़ंक्शन में, यदि सब कुछ ठीक रहता है तो सत्र प्रतिबद्ध होता है और यदि कोई अपवाद उठाया जाता है तो सत्र वापस ले लिया जाता है।

यह ध्यान रखना महत्वपूर्ण है कि यह कोड asyncio एक्सटेंशन के आसपास बनाया गया है। SQLAlchemy की यह सुविधा ऐप को डेटाबेस के साथ एसिंक्रोनस रूप से इंटरैक्ट करने की अनुमति देती है। इसका मतलब है कि डेटाबेस के अनुरोध अन्य एपीआई अनुरोधों को अवरुद्ध नहीं करेंगे, जिससे ऐप अधिक कुशल हो जाएगा।

एक बार मॉडल और कनेक्शन स्थापित हो जाने के बाद, अगला कदम यह सुनिश्चित करना है कि मॉडल डेटाबेस में जोड़े गए हैं।

त्वरित प्रवास

SQLAlchemy मॉडल डेटाबेस की संरचनाओं का प्रतिनिधित्व करते हैं। हालाँकि, केवल उन्हें बनाने से डेटाबेस में तत्काल परिवर्तन नहीं होता है। परिवर्तन करने के लिए, आपको पहले उन्हें लागू करना होगा. यह आमतौर पर एलेम्बिक जैसी माइग्रेशन लाइब्रेरी का उपयोग करके किया जाता है, जो हर मॉडल को ट्रैक करता है और तदनुसार डेटाबेस को अपडेट करता है।

चूँकि इस परिदृश्य में मॉडलों में कोई और बदलाव की योजना नहीं है, इसलिए एक बुनियादी माइग्रेशन स्क्रिप्ट पर्याप्त होगी। नीचे 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())

सीधे शब्दों में कहें तो, migrate_tables फ़ंक्शन मॉडल की संरचना को पढ़ता है और SQLAlchemy इंजन का उपयोग करके इसे डेटाबेस में फिर से बनाता है। इस स्क्रिप्ट को चलाने के लिए, python scripts/migrate.py कमांड का उपयोग करें।

मॉडल अब कोड और डेटाबेस दोनों में मौजूद हैं और get_db_session डेटाबेस के साथ इंटरैक्शन की सुविधा प्रदान कर सकते हैं। अब आप एपीआई लॉजिक पर काम करना शुरू कर सकते हैं।

ORM के साथ एपीआई

जैसा कि पहले उल्लेख किया गया है, सामग्री और औषधि के लिए एपीआई तीन कार्यों का समर्थन करने के लिए है:

• वस्तुएँ बनाना
• वस्तुओं को सूचीबद्ध करना
• आईडी द्वारा वस्तुओं को पुनः प्राप्त करना

पूर्व तैयारियों के लिए धन्यवाद, इन सभी सुविधाओं को पहले से ही ORM के रूप में SQLAlchemy और वेब फ्रेमवर्क के रूप में FastAPI के साथ लागू किया जा सकता है। आरंभ करने के लिए, alchemist/api/v1/routes.py फ़ाइल में स्थित सामग्री API की समीक्षा करें:

 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 एपीआई के अंतर्गत, तीन मार्ग उपलब्ध हैं। POST समापन बिंदु पहले से बनाए गए मॉडल और डेटाबेस सत्र से एक घटक पेलोड को एक ऑब्जेक्ट के रूप में लेता है। get_db_session जेनरेटर फ़ंक्शन सत्र प्रारंभ करता है और डेटाबेस इंटरैक्शन को सक्षम बनाता है।

वास्तविक कार्य निकाय में, पाँच चरण होते हैं:

1. आने वाले पेलोड से एक घटक वस्तु बनाई जाती है।
2. सत्र ऑब्जेक्ट की add विधि घटक ऑब्जेक्ट को सत्र ट्रैकिंग सिस्टम में जोड़ती है और इसे डेटाबेस में प्रविष्टि के लिए लंबित के रूप में चिह्नित करती है।
3. सत्र प्रतिबद्ध है.
4. यह सुनिश्चित करने के लिए घटक ऑब्जेक्ट को ताज़ा किया जाता है कि उसकी विशेषताएँ डेटाबेस स्थिति से मेल खाती हैं।
5. डेटाबेस घटक उदाहरण को from_orm विधि का उपयोग करके एपीआई मॉडल उदाहरण में परिवर्तित किया जाता है।

त्वरित परीक्षण के लिए, चल रहे ऐप के विरुद्ध एक सरल कर्ल निष्पादित किया जा सकता है:

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

प्रतिक्रिया में, एक घटक वस्तु होनी चाहिए जिसमें डेटाबेस से आने वाली एक आईडी हो:

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

हालाँकि SQLAlchemy की अमूर्तता की कई परतें एक साधारण API के लिए अनावश्यक लग सकती हैं, वे ORM विवरणों को अलग रखते हैं और SQLAlchemy की दक्षता और स्केलेबिलिटी में योगदान करते हैं। Asyncio के साथ संयुक्त होने पर, ORM सुविधाएँ API में असाधारण रूप से अच्छा प्रदर्शन करती हैं।

शेष दो समापन बिंदु कम जटिल हैं और समानताएं साझा करते हैं। एक हिस्सा जो गहन व्याख्या का हकदार है वह है get_ingredients फ़ंक्शन के अंदर scalars विधि का उपयोग। SQLAlchemy का उपयोग करके डेटाबेस को क्वेरी करते समय, execute विधि का उपयोग अक्सर तर्क के रूप में क्वेरी के साथ किया जाता है। जबकि execute विधि पंक्ति-जैसे टुपल्स लौटाती है, scalars सीधे ORM इकाइयाँ लौटाता है, जिससे समापन बिंदु साफ़ हो जाता है।

अब, उसी फ़ाइल में, पोशन एपीआई पर विचार करें:

 @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)

औषधि के लिए GET समापन बिंदु सामग्री के लिए समान हैं। हालाँकि, POST फ़ंक्शन को अतिरिक्त कोड की आवश्यकता होती है। ऐसा इसलिए है क्योंकि औषधि बनाने में कम से कम एक घटक आईडी शामिल होती है, जिसका अर्थ है कि सामग्री को लाया जाना चाहिए और नव निर्मित औषधि से जोड़ा जाना चाहिए। इसे प्राप्त करने के लिए, scalars विधि का फिर से उपयोग किया जाता है, लेकिन इस बार एक क्वेरी के साथ जो प्राप्त सामग्री की आईडी निर्दिष्ट करती है। औषधि निर्माण प्रक्रिया का शेष भाग सामग्री के समान है।

समापन बिंदु का परीक्षण करने के लिए, फिर से एक कर्ल कमांड निष्पादित किया जा सकता है।

 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"}'

इसके परिणामस्वरूप निम्नलिखित प्रतिक्रिया मिलती है:

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

यह ध्यान रखना महत्वपूर्ण है कि प्रत्येक घटक को औषधि के भीतर एक पूर्ण वस्तु के रूप में दर्शाया गया है, रिश्ते में निर्दिष्ट lazy="selectin" तर्क के लिए धन्यवाद।

एपीआई कार्यात्मक हैं, लेकिन कोड के साथ एक बड़ी समस्या है। जबकि SQLAlchemy आपको अपनी इच्छानुसार डेटाबेस के साथ इंटरैक्ट करने की स्वतंत्रता देता है, यह Django के Model.objects के समान कोई उच्च-स्तरीय "प्रबंधक" उपयोगिता प्रदान नहीं करता है। परिणामस्वरूप, आपको इसे स्वयं बनाने की आवश्यकता होगी, जो अनिवार्य रूप से घटक और औषधि एपीआई में उपयोग किया जाने वाला तर्क है। हालाँकि, यदि आप इस तर्क को एक अलग स्थान पर निकाले बिना सीधे अंतिम बिंदुओं में रखते हैं, तो आपके पास बहुत सारे डुप्लिकेट कोड होंगे। इसके अतिरिक्त, प्रश्नों या मॉडलों में परिवर्तन करना प्रबंधित करना कठिन हो जाएगा।

आगामी अध्याय रिपोजिटरी पैटर्न का परिचय देता है: ओआरएम कोड निकालने के लिए एक सुंदर समाधान।

कोष

रिपॉजिटरी पैटर्न डेटाबेस के साथ काम करने के विवरण को अलग करने की अनुमति देता है। SQLAlchemy का उपयोग करने के मामले में, जैसे कि अल्केमिस्ट के उदाहरण में, रिपॉजिटरी क्लास कई मॉडलों को प्रबंधित करने और डेटाबेस सत्र के साथ इंटरैक्ट करने के लिए जिम्मेदार होगा।

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))

DatabaseRepository वर्ग सभी तर्क रखता है जो पहले समापन बिंदुओं में शामिल थे। अंतर यह है कि यह विशिष्ट मॉडल वर्ग को __init__ विधि में पारित करने की अनुमति देता है, जिससे प्रत्येक एंडपॉइंट में डुप्लिकेट करने के बजाय सभी मॉडलों के लिए कोड का पुन: उपयोग करना संभव हो जाता है।

इसके अलावा, DatabaseRepository Python जेनेरिक का उपयोग करता है, जिसमें Model जेनेरिक प्रकार अमूर्त डेटाबेस मॉडल से जुड़ा होता है। यह रिपॉजिटरी वर्ग को स्थैतिक प्रकार की जाँच से अधिक लाभ उठाने की अनुमति देता है। जब किसी विशिष्ट मॉडल के साथ उपयोग किया जाता है, तो रिपॉजिटरी विधियों के रिटर्न प्रकार इस विशिष्ट मॉडल को प्रतिबिंबित करेंगे।

क्योंकि रिपॉजिटरी को डेटाबेस सत्र का उपयोग करने की आवश्यकता होती है, इसे 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

सीधे शब्दों में कहें तो get_repository फ़ंक्शन एक निर्भरता फ़ैक्टरी है। यह सबसे पहले डेटाबेस मॉडल लेता है जिसके साथ आप रिपॉजिटरी का उपयोग करेंगे। फिर, यह निर्भरता लौटाता है, जिसका उपयोग डेटाबेस सत्र प्राप्त करने और रिपॉजिटरी ऑब्जेक्ट को प्रारंभ करने के लिए किया जाएगा। बेहतर समझ हासिल करने के लिए, alchemist/api/v2/routes.py फ़ाइल से नई API देखें। यह केवल POST समापन बिंदु दिखाता है, लेकिन यह आपको यह स्पष्ट विचार देने के लिए पर्याप्त होना चाहिए कि कोड में सुधार कैसे होता है:

 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)

ध्यान देने योग्य पहली महत्वपूर्ण विशेषता Annotated का उपयोग है, जो फास्टएपीआई निर्भरता के साथ काम करने का एक नया तरीका है। निर्भरता के रिटर्न प्रकार को DatabaseRepository[db_models.Ingredient] के रूप में निर्दिष्ट करके और Depends(get_repository(db_models.Ingredient)) के साथ इसके उपयोग की घोषणा करके आप एंडपॉइंट में सरल प्रकार के एनोटेशन का उपयोग कर सकते हैं: repository: IngredientRepository ।

रिपॉजिटरी के लिए धन्यवाद, एंडपॉइंट्स को सभी ORM-संबंधित बोझ को संग्रहीत करने की आवश्यकता नहीं है। यहां तक कि अधिक जटिल औषधि मामले में भी, आपको बस एक ही समय में दो रिपॉजिटरी का उपयोग करना होगा।

आपको आश्चर्य हो सकता है कि क्या दो रिपॉजिटरी प्रारंभ करने से सत्र दो बार प्रारंभ होगा। जवाब न है। फास्टएपीआई निर्भरता प्रणाली एक ही अनुरोध में समान निर्भरता कॉल को कैश करती है। इसका मतलब यह है कि सत्र आरंभीकरण कैश हो जाता है और दोनों रिपॉजिटरी ठीक उसी सत्र ऑब्जेक्ट का उपयोग करते हैं। SQLAlchemy और FastAPI के संयोजन की एक और बड़ी विशेषता।

एपीआई पूरी तरह कार्यात्मक है और इसमें पुन: प्रयोज्य, उच्च प्रदर्शन वाली डेटा-एक्सेस परत है। अगला कदम यह सुनिश्चित करना है कि कुछ एंड-टू-एंड परीक्षण लिखकर आवश्यकताओं को पूरा किया जाए।

परिक्षण

सॉफ़्टवेयर विकास में परीक्षण महत्वपूर्ण भूमिका निभाते हैं। परियोजनाओं में इकाई, एकीकरण और एंड-टू-एंड (ई2ई) परीक्षण शामिल हो सकते हैं। हालांकि आमतौर पर अधिक संख्या में सार्थक इकाई परीक्षण करना सबसे अच्छा होता है, यह सुनिश्चित करने के लिए कि संपूर्ण वर्कफ़्लो सही ढंग से काम कर रहा है, कम से कम कुछ E2E परीक्षण लिखना भी अच्छा है।

अल्केमिस्ट ऐप के लिए कुछ E2E परीक्षण बनाने के लिए, दो अतिरिक्त लाइब्रेरी की आवश्यकता होती है:

• वास्तव में परीक्षण बनाने और चलाने के लिए पाइटेस्ट
• परीक्षणों के अंदर एसिंक अनुरोध करने के लिए httpx

एक बार ये स्थापित हो जाने के बाद, अगला कदम एक अलग, परीक्षण डेटाबेस स्थापित करना है। आप नहीं चाहते कि आपका डिफ़ॉल्ट डेटाबेस प्रदूषित हो या ख़त्म हो जाए। चूंकि अल्केमिस्ट में डॉकर सेटअप शामिल है, इसलिए दूसरा डेटाबेस बनाने के लिए केवल एक सरल स्क्रिप्ट की आवश्यकता होती है। scripts/create_test_db.sh फ़ाइल से कोड पर एक नज़र डालें:

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

स्क्रिप्ट को निष्पादित करने के लिए, इसे पोस्टग्रेज कंटेनर में वॉल्यूम के रूप में जोड़ा जाना चाहिए। इसे docker-compose.yaml फ़ाइल के volumes अनुभाग में शामिल करके प्राप्त किया जा सकता है।

तैयारी का अंतिम चरण 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

एक चीज़ जो परीक्षणों में बदलना आवश्यक है, वह यह है कि ऐप डेटाबेस के साथ कैसे इंटरैक्ट करता है। इसमें न केवल डेटाबेस यूआरएल को बदलना शामिल है बल्कि यह सुनिश्चित करना भी शामिल है कि प्रत्येक परीक्षण एक खाली डेटाबेस से शुरू करके अलग किया गया है।

db_session फिक्स्चर इन दोनों लक्ष्यों को पूरा करता है। इसका शरीर निम्नलिखित कदम उठाता है:

1. संशोधित डेटाबेस URL के साथ एक इंजन बनाएं।
2. यह सुनिश्चित करने के लिए कि परीक्षण में एक साफ़ डेटाबेस है, सभी मौजूदा तालिकाएँ हटा दें।
3. डेटाबेस के अंदर सभी तालिकाएँ बनाएँ (माइग्रेशन स्क्रिप्ट के समान कोड)।
4. एक सत्र ऑब्जेक्ट बनाएं और उत्पन्न करें।
5. परीक्षण पूरा हो जाने पर सत्र को मैन्युअल रूप से बंद कर दें।

हालाँकि अंतिम चरण को संदर्भ प्रबंधक के रूप में भी लागू किया जा सकता है, इस मामले में मैन्युअल समापन ठीक काम करता है।

शेष दो फिक्स्चर काफी आत्म-व्याख्यात्मक होने चाहिए:

• test_app alchemist/app.py फ़ाइल से फास्टएपीआई उदाहरण है, जिसमें get_db_session निर्भरता को db_session फिक्स्चर से बदल दिया गया है
• client httpx AsyncClient है जो test_app के विरुद्ध API अनुरोध करेगा

यह सब स्थापित होने के बाद, अंततः वास्तविक परीक्षण लिखे जा सकते हैं। संक्षिप्तता के लिए, 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

परीक्षण फिक्स्चर में बनाए गए क्लाइंट ऑब्जेक्ट का उपयोग करता है, जो ओवरराइड निर्भरता के साथ फास्टएपीआई उदाहरण के लिए अनुरोध करता है। परिणामस्वरूप, परीक्षण एक अलग डेटाबेस के साथ इंटरैक्ट करने में सक्षम है जिसे परीक्षण पूरा होने के बाद साफ़ कर दिया जाएगा। दोनों एपीआई के लिए शेष परीक्षण सूट की संरचना काफी हद तक समान है।

सारांश

आधुनिक और शक्तिशाली बैकएंड एप्लिकेशन बनाने के लिए फास्टएपीआई और एसक्यूएलकेमी उत्कृष्ट तकनीकें हैं। वे जो स्वतंत्रता, सरलता और लचीलापन प्रदान करते हैं, वह उन्हें पायथन-आधारित परियोजनाओं के लिए सर्वोत्तम विकल्पों में से एक बनाता है। यदि डेवलपर्स सर्वोत्तम प्रथाओं और पैटर्न का पालन करते हैं, तो वे प्रदर्शन करने वाले, मजबूत और अच्छी तरह से संरचित एप्लिकेशन बना सकते हैं जो डेटाबेस संचालन और एपीआई तर्क को आसानी से संभालते हैं। इस लेख का उद्देश्य आपको इस अद्भुत संयोजन को स्थापित करने और बनाए रखने की अच्छी समझ प्रदान करना है।

सूत्रों का कहना है

अल्केमिस्ट परियोजना का स्रोत कोड यहां पाया जा सकता है: लिंक ।