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

जबकि Django और फ्लास्क कई पायथन इंजीनियरों के लिए पहली पसंद बने हुए हैं, पहले से ही एक निर्विवाद रूप से विश्वसनीय पसंद के रूप में मान्यता दी गई है। यह एक अत्यधिक लचीला, अच्छी तरह से अनुकूलित, संरचित ढांचा है जो डेवलपर को बैकएंड एप्लिकेशन बनाने की अनंत संभावनाएं देता है। फास्टएपीआई को डेटाबेस के साथ काम करना अधिकांश बैकएंड अनुप्रयोगों का एक अनिवार्य पहलू है। परिणामस्वरूप, ORM बैकएंड कोड में एक महत्वपूर्ण भूमिका निभाता है। हालाँकि, Django के विपरीत, FastAPI में ORM बिल्ट-इन नहीं है। एक उपयुक्त लाइब्रेरी का चयन करना और उसे कोडबेस में एकीकृत करना पूरी तरह से डेवलपर की जिम्मेदारी है। पायथन इंजीनियर व्यापक रूप से को सबसे लोकप्रिय उपलब्ध ORM मानते हैं। यह एक प्रसिद्ध पुस्तकालय है जिसका उपयोग 2006 से किया जा रहा है और इसे हजारों परियोजनाओं द्वारा अपनाया गया है। 2023 में, इसे संस्करण 2.0 में एक बड़ा अपडेट प्राप्त हुआ। फास्टएपीआई के समान, SQLAlchemy डेवलपर्स को शक्तिशाली सुविधाओं और उपयोगिताओं को एक विशिष्ट तरीके से उपयोग करने के लिए मजबूर किए बिना प्रदान करता है। अनिवार्य रूप से, यह एक बहुमुखी टूलकिट है जो डेवलपर्स को इसे अपनी इच्छानुसार उपयोग करने का अधिकार देता है। 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 में उत्पादन चरण प्रवासन के लिए एलेम्बिक सेटअप परीक्षणों के लिए उपनिर्देशिकाएँ यह जानबूझकर जटिलता को कम करने और अनावश्यक ओवरहेड से बचने के लिए किया गया था। हालाँकि, अधिक उत्पादन-तैयार परियोजना के साथ काम करते समय इन कारकों को ध्यान में रखना महत्वपूर्ण है। एपीआई आवश्यकताएँ किसी ऐप को विकसित करना शुरू करते समय, उन मॉडलों पर विचार करना महत्वपूर्ण है जिनका उपयोग आपका ऐप करेगा। ये मॉडल उन और का प्रतिनिधित्व करेंगे जिनके साथ आपका ऐप काम करेगा और एपीआई में प्रदर्शित किया जाएगा। अल्केमिस्ट ऐप के मामले में, दो इकाइयाँ हैं: सामग्री और औषधि। एपीआई को इन संस्थाओं को बनाने और पुनः प्राप्त करने की अनुमति देनी चाहिए। फ़ाइल में वे मॉडल शामिल हैं जिनका उपयोग API में किया जाएगा: वस्तुओं संस्थाओं 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) एपीआई और मॉडल लौटाएगा। कॉन्फ़िगरेशन में को पर सेट करने से भविष्य में SQLAlchemy ऑब्जेक्ट के साथ काम करना आसान हो जाएगा। नई वस्तुएं बनाने के लिए मॉडल का उपयोग किया जाएगा। Ingredient Potion orm_mode True 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 मॉडल पर आगे बढ़ते हुए, विशेषता डेटाबेस तालिका का नाम निर्दिष्ट करती है, जबकि फ़ील्ड नए SQLAlchemy सिंटैक्स का उपयोग करती है, जिससे मॉडल फ़ील्ड को प्रकार एनोटेशन के साथ घोषित किया जा सकता है। यह संक्षिप्त और आधुनिक दृष्टिकोण टाइप चेकर्स और आईडीई के लिए शक्तिशाली और लाभप्रद दोनों है, क्योंकि यह फ़ील्ड को एक स्ट्रिंग के रूप में पहचानता है। Ingredient __tablename__ name name मॉडल में चीजें अधिक जटिल हो जाती हैं। इसमें और विशेषताएँ भी शामिल हैं, लेकिन इसके अलावा, यह अवयवों से संबंध संग्रहीत करता है। का उपयोग इंगित करता है कि औषधि में कई सामग्रियां हो सकती हैं, और इस मामले में, संबंध अनेक-से-अनेक (एम2एम) है। इसका मतलब यह है कि एक ही घटक को कई औषधियों में मिलाया जा सकता है। Potion __tablename__ name Mapped[list["Ingredient"]] एम2एम को अतिरिक्त कॉन्फ़िगरेशन की आवश्यकता होती है, जिसमें आमतौर पर एक एसोसिएशन तालिका का निर्माण शामिल होता है जो दो संस्थाओं के बीच कनेक्शन को संग्रहीत करता है। इस मामले में, ऑब्जेक्ट केवल घटक और पोशन के पहचानकर्ताओं को संग्रहीत करता है, लेकिन इसमें अतिरिक्त गुण भी शामिल हो सकते हैं, जैसे कि पोशन के लिए आवश्यक विशिष्ट घटक की मात्रा। potion_ingredient_association फ़ंक्शन औषधि और उसके अवयवों के बीच संबंध को कॉन्फ़िगर करता है। तर्क निर्दिष्ट करता है कि संबंधित वस्तुओं को कैसे लोड किया जाना चाहिए। दूसरे शब्दों में: जब आप औषधि ला रहे हों तो SQLAlchemy को संबंधित सामग्रियों के साथ क्या करना चाहिए। इसे पर सेट करने का मतलब है कि अवयवों को पोशन के साथ लोड किया जाएगा, जिससे कोड में अतिरिक्त प्रश्नों की आवश्यकता समाप्त हो जाएगी। relationship lazy selectin ORM के साथ काम करते समय अच्छी तरह से डिज़ाइन किए गए मॉडल बनाना महत्वपूर्ण है। एक बार यह हो जाने के बाद, अगला चरण डेटाबेस के साथ कनेक्शन स्थापित करना है। सत्र संचालक डेटाबेस के साथ काम करते समय, विशेष रूप से SQLAlchemy का उपयोग करते समय, निम्नलिखित अवधारणाओं को समझना आवश्यक है: बोली इंजन कनेक्शन कनेक्शन पूल सत्र इन सभी शब्दों में से सबसे महत्वपूर्ण है । SQLAlchemy दस्तावेज़ के अनुसार, इंजन ऑब्जेक्ट डेटाबेस कनेक्टिविटी और व्यवहार को सुविधाजनक बनाने के लिए और जोड़ने के लिए ज़िम्मेदार है। सरल शब्दों में, इंजन ऑब्जेक्ट डेटाबेस कनेक्शन का स्रोत है, जबकि SQL स्टेटमेंट निष्पादित करने, लेनदेन प्रबंधित करने और डेटाबेस से परिणाम पुनर्प्राप्त करने जैसी उच्च-स्तरीय कार्यक्षमता प्रदान करता है। इंजन Pool Dialect कनेक्शन एक कार्य की एक इकाई है जो एक ही लेन-देन के भीतर संबंधित कार्यों को समूहित करता है। यह अंतर्निहित डेटाबेस कनेक्शन पर एक अमूर्त है और कुशलतापूर्वक कनेक्शन और लेनदेन व्यवहार का प्रबंधन करता है। सत्र एक घटक है जो एक विशिष्ट डेटाबेस बैकएंड के लिए समर्थन प्रदान करता है। यह 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()) सीधे शब्दों में कहें तो, फ़ंक्शन मॉडल की संरचना को पढ़ता है और SQLAlchemy इंजन का उपयोग करके इसे डेटाबेस में फिर से बनाता है। इस स्क्रिप्ट को चलाने के लिए, कमांड का उपयोग करें। migrate_tables python scripts/migrate.py मॉडल अब कोड और डेटाबेस दोनों में मौजूद हैं और डेटाबेस के साथ इंटरैक्शन की सुविधा प्रदान कर सकते हैं। अब आप एपीआई लॉजिक पर काम करना शुरू कर सकते हैं। get_db_session ORM के साथ एपीआई जैसा कि पहले उल्लेख किया गया है, सामग्री और औषधि के लिए एपीआई तीन कार्यों का समर्थन करने के लिए है: वस्तुएँ बनाना वस्तुओं को सूचीबद्ध करना आईडी द्वारा वस्तुओं को पुनः प्राप्त करना पूर्व तैयारियों के लिए धन्यवाद, इन सभी सुविधाओं को पहले से ही ORM के रूप में SQLAlchemy और वेब फ्रेमवर्क के रूप में FastAPI के साथ लागू किया जा सकता है। आरंभ करने के लिए, फ़ाइल में स्थित सामग्री API की समीक्षा करें: 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) एपीआई के अंतर्गत, तीन मार्ग उपलब्ध हैं। POST समापन बिंदु पहले से बनाए गए मॉडल और डेटाबेस सत्र से एक घटक पेलोड को एक ऑब्जेक्ट के रूप में लेता है। जेनरेटर फ़ंक्शन सत्र प्रारंभ करता है और डेटाबेस इंटरैक्शन को सक्षम बनाता है। /ingredients get_db_session वास्तविक कार्य निकाय में, पाँच चरण होते हैं: आने वाले पेलोड से एक घटक वस्तु बनाई जाती है। सत्र ऑब्जेक्ट की विधि घटक ऑब्जेक्ट को सत्र ट्रैकिंग सिस्टम में जोड़ती है और इसे डेटाबेस में प्रविष्टि के लिए लंबित के रूप में चिह्नित करती है। add सत्र प्रतिबद्ध है. यह सुनिश्चित करने के लिए घटक ऑब्जेक्ट को ताज़ा किया जाता है कि उसकी विशेषताएँ डेटाबेस स्थिति से मेल खाती हैं। डेटाबेस घटक उदाहरण को विधि का उपयोग करके एपीआई मॉडल उदाहरण में परिवर्तित किया जाता है। 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 में असाधारण रूप से अच्छा प्रदर्शन करती हैं। शेष दो समापन बिंदु कम जटिल हैं और समानताएं साझा करते हैं। एक हिस्सा जो गहन व्याख्या का हकदार है वह है फ़ंक्शन के अंदर विधि का उपयोग। SQLAlchemy का उपयोग करके डेटाबेस को क्वेरी करते समय, विधि का उपयोग अक्सर तर्क के रूप में क्वेरी के साथ किया जाता है। जबकि विधि पंक्ति-जैसे टुपल्स लौटाती है, सीधे ORM इकाइयाँ लौटाता है, जिससे समापन बिंदु साफ़ हो जाता है। get_ingredients scalars execute execute scalars अब, उसी फ़ाइल में, पोशन एपीआई पर विचार करें: @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__ इसके अलावा, Python जेनेरिक का उपयोग करता है, जिसमें जेनेरिक प्रकार अमूर्त डेटाबेस मॉडल से जुड़ा होता है। यह रिपॉजिटरी वर्ग को स्थैतिक प्रकार की जाँच से अधिक लाभ उठाने की अनुमति देता है। जब किसी विशिष्ट मॉडल के साथ उपयोग किया जाता है, तो रिपॉजिटरी विधियों के रिटर्न प्रकार इस विशिष्ट मॉडल को प्रतिबिंबित करेंगे। DatabaseRepository 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 सीधे शब्दों में कहें तो फ़ंक्शन एक निर्भरता फ़ैक्टरी है। यह सबसे पहले डेटाबेस मॉडल लेता है जिसके साथ आप रिपॉजिटरी का उपयोग करेंगे। फिर, यह निर्भरता लौटाता है, जिसका उपयोग डेटाबेस सत्र प्राप्त करने और रिपॉजिटरी ऑब्जेक्ट को प्रारंभ करने के लिए किया जाएगा। बेहतर समझ हासिल करने के लिए, फ़ाइल से नई API देखें। यह केवल POST समापन बिंदु दिखाता है, लेकिन यह आपको यह स्पष्ट विचार देने के लिए पर्याप्त होना चाहिए कि कोड में सुधार कैसे होता है: 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) ध्यान देने योग्य पहली महत्वपूर्ण विशेषता का उपयोग है, जो फास्टएपीआई निर्भरता के साथ काम करने का एक नया तरीका है। निर्भरता के रिटर्न प्रकार को के रूप में निर्दिष्ट करके और के साथ इसके उपयोग की घोषणा करके आप एंडपॉइंट में सरल प्रकार के एनोटेशन का उपयोग कर सकते हैं: । 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 संशोधित डेटाबेस URL के साथ एक इंजन बनाएं। यह सुनिश्चित करने के लिए कि परीक्षण में एक साफ़ डेटाबेस है, सभी मौजूदा तालिकाएँ हटा दें। डेटाबेस के अंदर सभी तालिकाएँ बनाएँ (माइग्रेशन स्क्रिप्ट के समान कोड)। एक सत्र ऑब्जेक्ट बनाएं और उत्पन्न करें। परीक्षण पूरा हो जाने पर सत्र को मैन्युअल रूप से बंद कर दें। हालाँकि अंतिम चरण को संदर्भ प्रबंधक के रूप में भी लागू किया जा सकता है, इस मामले में मैन्युअल समापन ठीक काम करता है। शेष दो फिक्स्चर काफी आत्म-व्याख्यात्मक होने चाहिए: फ़ाइल से फास्टएपीआई उदाहरण है, जिसमें निर्भरता को फिक्स्चर से बदल दिया गया है test_app alchemist/app.py get_db_session db_session httpx है जो के विरुद्ध API अनुरोध करेगा client AsyncClient test_app यह सब स्थापित होने के बाद, अंततः वास्तविक परीक्षण लिखे जा सकते हैं। संक्षिप्तता के लिए, फ़ाइल से नीचे दिया गया उदाहरण केवल एक घटक बनाने के लिए एक परीक्षण दिखाता है: 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 परीक्षण फिक्स्चर में बनाए गए क्लाइंट ऑब्जेक्ट का उपयोग करता है, जो ओवरराइड निर्भरता के साथ फास्टएपीआई उदाहरण के लिए अनुरोध करता है। परिणामस्वरूप, परीक्षण एक अलग डेटाबेस के साथ इंटरैक्ट करने में सक्षम है जिसे परीक्षण पूरा होने के बाद साफ़ कर दिया जाएगा। दोनों एपीआई के लिए शेष परीक्षण सूट की संरचना काफी हद तक समान है। सारांश आधुनिक और शक्तिशाली बैकएंड एप्लिकेशन बनाने के लिए फास्टएपीआई और एसक्यूएलकेमी उत्कृष्ट तकनीकें हैं। वे जो स्वतंत्रता, सरलता और लचीलापन प्रदान करते हैं, वह उन्हें पायथन-आधारित परियोजनाओं के लिए सर्वोत्तम विकल्पों में से एक बनाता है। यदि डेवलपर्स सर्वोत्तम प्रथाओं और पैटर्न का पालन करते हैं, तो वे प्रदर्शन करने वाले, मजबूत और अच्छी तरह से संरचित एप्लिकेशन बना सकते हैं जो डेटाबेस संचालन और एपीआई तर्क को आसानी से संभालते हैं। इस लेख का उद्देश्य आपको इस अद्भुत संयोजन को स्थापित करने और बनाए रखने की अच्छी समझ प्रदान करना है। सूत्रों का कहना है अल्केमिस्ट परियोजना का स्रोत कोड यहां पाया जा सकता है: । लिंक भी प्रकाशित किया गया है. यहाँ