ফাস্টএপিআই-এর সাথে SQLAlchemy 2.0 ব্যবহার করার জন্য নিদর্শন এবং অনুশীলন

যদিও জ্যাঙ্গো এবং ফ্লাস্ক অনেক পাইথন ইঞ্জিনিয়ারদের জন্য প্রথম পছন্দ হিসাবে রয়ে গেছে, ফাস্টএপিআই ইতিমধ্যেই একটি অনস্বীকার্যভাবে নির্ভরযোগ্য বাছাই হিসাবে স্বীকৃত হয়েছে। এটি একটি অত্যন্ত নমনীয়, ভাল-অপ্টিমাইজ করা, কাঠামোবদ্ধ কাঠামো যা বিকাশকারীকে ব্যাকএন্ড অ্যাপ্লিকেশন তৈরির জন্য অফুরন্ত সম্ভাবনা দেয়।

ডাটাবেসের সাথে কাজ করা বেশিরভাগ ব্যাকএন্ড অ্যাপ্লিকেশনের একটি অপরিহার্য দিক। ফলস্বরূপ, ORM ব্যাকএন্ড কোডে একটি গুরুত্বপূর্ণ ভূমিকা পালন করে। যাইহোক, জ্যাঙ্গোর বিপরীতে, ফাস্টএপিআইতে একটি ORM বিল্ট-ইন নেই। একটি উপযুক্ত লাইব্রেরি নির্বাচন করা এবং এটিকে কোডবেসে সংহত করা সম্পূর্ণরূপে ডেভেলপারের দায়িত্ব।

পাইথন ইঞ্জিনিয়াররা ব্যাপকভাবে SQLAlchemy-কে সবচেয়ে জনপ্রিয় ওআরএম উপলব্ধ বলে মনে করেন। এটি একটি কিংবদন্তি গ্রন্থাগার যা 2006 সাল থেকে ব্যবহার করা হচ্ছে এবং হাজার হাজার প্রকল্প দ্বারা গৃহীত হয়েছে। 2023 সালে, এটি 2.0 সংস্করণে একটি বড় আপডেট পেয়েছে। FastAPI-এর মতো, SQLAlchemy ডেভেলপারদের একটি নির্দিষ্ট উপায়ে ব্যবহার করতে বাধ্য না করে শক্তিশালী বৈশিষ্ট্য এবং ইউটিলিটি প্রদান করে। মূলত, এটি একটি বহুমুখী টুলকিট যা ডেভেলপারদের উপযুক্ত মনে হলেও এটি ব্যবহার করার ক্ষমতা দেয়।

FastAPI এবং SQLAlchemy স্বর্গে তৈরি একটি মিল। তারা উভয়ই নির্ভরযোগ্য, পারফরম্যান্স এবং আধুনিক প্রযুক্তি, যা শক্তিশালী এবং অনন্য অ্যাপ্লিকেশন তৈরি করতে সক্ষম করে। এই নিবন্ধটি একটি FastAPI ব্যাকএন্ড অ্যাপ্লিকেশন তৈরি করে যা SQLAlchemy 2.0 কে ORM হিসাবে ব্যবহার করে। বিষয়বস্তু কভার করে:

• Mapped এবং mapped_column ব্যবহার করে মডেল তৈরি করুন
• একটি বিমূর্ত মডেল সংজ্ঞায়িত করা
• ডাটাবেস সেশন পরিচালনা
• ORM ব্যবহার করে
• সমস্ত মডেলের জন্য একটি সাধারণ সংগ্রহস্থল শ্রেণী তৈরি করা
• একটি পরীক্ষা সেটআপ প্রস্তুত করা এবং পরীক্ষা যোগ করা

এর পরে, আপনি সহজেই SQLAlchemy ORM-এর সাথে FastAPI অ্যাপ্লিকেশনটি একত্রিত করতে সক্ষম হবেন। উপরন্তু, আপনি সুগঠিত, মজবুত এবং পারফরম্যান্স অ্যাপ্লিকেশন তৈরি করার জন্য সর্বোত্তম অনুশীলন এবং নিদর্শনগুলির সাথে পরিচিত হবেন।

পূর্বশর্ত

নিবন্ধে অন্তর্ভুক্ত কোড উদাহরণগুলি অ্যালকেমিস্ট প্রকল্প থেকে এসেছে, যা উপাদান এবং ওষুধের বস্তু তৈরি এবং পড়ার জন্য একটি মৌলিক API। নিবন্ধটির মূল ফোকাস হল FastAPI এবং SQLAlchemy এর সমন্বয় অন্বেষণ করা। এটি অন্যান্য বিষয়গুলিকে কভার করে না, যেমন:

• ডকার সেটআপ কনফিগার করা হচ্ছে
• ইউভিকর্ন সার্ভার শুরু হচ্ছে
• লিন্টিং স্থাপন করা

আপনি যদি এই বিষয়গুলিতে আগ্রহী হন তবে আপনি কোডবেস পরীক্ষা করে নিজেরাই সেগুলি অন্বেষণ করতে পারেন। অ্যালকেমিস্ট প্রকল্পের কোড সংগ্রহস্থল অ্যাক্সেস করতে, এখানে এই লিঙ্কটি অনুসরণ করুন। উপরন্তু, আপনি নীচের প্রকল্পের ফাইল গঠন খুঁজে পেতে পারেন:

 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

যদিও গাছটি বড় মনে হতে পারে, তবে কিছু বিষয়বস্তু এই নিবন্ধের মূল বিষয়ের সাথে প্রাসঙ্গিক নয়। উপরন্তু, কিছু নির্দিষ্ট এলাকায় যা প্রয়োজন তার থেকে কোডটি সহজ প্রদর্শিত হতে পারে। উদাহরণস্বরূপ, প্রকল্পের অভাব রয়েছে:

• ডকারফাইলে উৎপাদন পর্যায়
• মাইগ্রেশনের জন্য অ্যালেম্বিক সেটআপ
• পরীক্ষার জন্য সাবডিরেক্টরি

জটিলতা কমাতে এবং অপ্রয়োজনীয় ওভারহেড এড়াতে এটি ইচ্ছাকৃতভাবে করা হয়েছিল। যাইহোক, আরও উত্পাদন-প্রস্তুত প্রকল্পের সাথে কাজ করার সময় এই বিষয়গুলি মাথায় রাখা গুরুত্বপূর্ণ।

API প্রয়োজনীয়তা

একটি অ্যাপ বিকাশ শুরু করার সময়, আপনার অ্যাপ ব্যবহার করবে এমন মডেলগুলি বিবেচনা করা গুরুত্বপূর্ণ। এই মডেলগুলি সেই বস্তু এবং সত্তাগুলিকে প্রতিনিধিত্ব করবে যেগুলির সাথে আপনার অ্যাপটি কাজ করবে এবং API-তে প্রকাশ করা হবে৷ অ্যালকেমিস্ট অ্যাপের ক্ষেত্রে, দুটি সত্তা রয়েছে: উপাদান এবং ওষুধ। API-কে এই সত্তাগুলি তৈরি এবং পুনরুদ্ধার করার অনুমতি দেওয়া উচিত। 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 মডেল ব্যবহার করা হবে।

pydantic এর ব্যবহার ক্লাসগুলিকে তাদের ভূমিকা এবং ফাংশনে আরও বিস্তারিত এবং স্পষ্ট করে তোলে। এখন, ডাটাবেস মডেল তৈরি করার সময়।

মডেল ঘোষণা

একটি মডেল মূলত কিছু একটি প্রতিনিধিত্ব . এপিআই-এর প্রেক্ষাপটে, মডেলগুলি রিকোয়েস্ট বডিতে ব্যাকএন্ড কী প্রত্যাশা করে এবং প্রতিক্রিয়া ডেটাতে কী ফিরে আসবে তা উপস্থাপন করে। ডাটাবেস মডেল, অন্যদিকে, আরও জটিল এবং ডাটাবেসে সংরক্ষিত ডেটা স্ট্রাকচার এবং তাদের মধ্যে সম্পর্কের ধরনগুলিকে উপস্থাপন করে।

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 সিনট্যাক্স ব্যবহার করে, মডেল ক্ষেত্রগুলিকে টাইপ টীকা সহ ঘোষণা করার অনুমতি দেয়। এই সংক্ষিপ্ত এবং আধুনিক পদ্ধতিটি টাইপ চেকার এবং IDE-এর জন্য শক্তিশালী এবং সুবিধাজনক, কারণ এটি name ক্ষেত্রটিকে একটি স্ট্রিং হিসাবে স্বীকৃতি দেয়।

Potion মডেলে জিনিসগুলি আরও জটিল হয়ে ওঠে। এটিতে __tablename__ এবং name বৈশিষ্ট্যগুলিও রয়েছে, কিন্তু তার উপরে, এটি উপাদানগুলির সাথে সম্পর্ক সংরক্ষণ করে। Mapped[list["Ingredient"]] ব্যবহার ইঙ্গিত করে যে ওষুধে একাধিক উপাদান থাকতে পারে এবং এই ক্ষেত্রে, সম্পর্কটি বহু-থেকে-অনেক (M2M)। এর মানে হল যে একটি একক উপাদান একাধিক ওষুধের জন্য বরাদ্দ করা যেতে পারে।

M2M-এর জন্য অতিরিক্ত কনফিগারেশন প্রয়োজন, সাধারণত একটি অ্যাসোসিয়েশন টেবিল তৈরি করা হয় যা দুটি সত্তার মধ্যে সংযোগ সংরক্ষণ করে। এই ক্ষেত্রে, 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-এর এই বৈশিষ্ট্যটি অ্যাপটিকে ডাটাবেসের সাথে অ্যাসিঙ্ক্রোনাসভাবে ইন্টারঅ্যাক্ট করতে দেয়। এর মানে হল যে ডাটাবেসের অনুরোধগুলি অন্যান্য API অনুরোধগুলিকে ব্লক করবে না, অ্যাপটিকে আরও দক্ষ করে তুলবে।

একবার মডেল এবং সংযোগ সেট আপ হয়ে গেলে, পরবর্তী ধাপ হল নিশ্চিত করা যে মডেলগুলি ডাটাবেসে যোগ করা হয়েছে।

দ্রুত মাইগ্রেশন

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 ডাটাবেসের সাথে মিথস্ক্রিয়াকে সহজতর করতে পারে। আপনি এখন API যুক্তিতে কাজ শুরু করতে পারেন।

ORM সহ API

পূর্বে উল্লিখিত হিসাবে, উপাদান এবং ওষুধের জন্য API তিনটি ক্রিয়াকলাপকে সমর্থন করার জন্য বোঝানো হয়েছে:

• বস্তু তৈরি করা
• বস্তু তালিকাভুক্ত করা
• আইডি দ্বারা বস্তু পুনরুদ্ধার করা হচ্ছে

পূর্ব প্রস্তুতির জন্য ধন্যবাদ, এই সমস্ত বৈশিষ্ট্যগুলি ইতিমধ্যেই SQLAlchemy-এর সাথে ORM এবং 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 API-এর অধীনে, তিনটি রুট উপলব্ধ রয়েছে। POST এন্ডপয়েন্ট পূর্বে তৈরি মডেল এবং একটি ডাটাবেস সেশন থেকে একটি বস্তু হিসাবে একটি উপাদান পেলোড নেয়। get_db_session জেনারেটর ফাংশন সেশন শুরু করে এবং ডাটাবেস ইন্টারঅ্যাকশন সক্ষম করে।

প্রকৃত ফাংশন বডিতে, পাঁচটি ধাপ হচ্ছে:

1. ইনকামিং পেলোড থেকে একটি উপাদান বস্তু তৈরি করা হয়।
2. সেশন অবজেক্টের add পদ্ধতি সেশন ট্র্যাকিং সিস্টেমে উপাদান বস্তু যোগ করে এবং ডাটাবেসে সন্নিবেশের জন্য মুলতুবি হিসাবে চিহ্নিত করে।
3. অধিবেশন প্রতিশ্রুতিবদ্ধ.
4. উপাদান বস্তুর বৈশিষ্ট্যগুলি ডাটাবেস অবস্থার সাথে মেলে তা নিশ্চিত করতে রিফ্রেশ করা হয়।
5. ডাটাবেস উপাদান উদাহরণ from_orm পদ্ধতি ব্যবহার করে API মডেল উদাহরণে রূপান্তরিত হয়।

একটি দ্রুত পরীক্ষার জন্য, চলমান অ্যাপের বিরুদ্ধে একটি সাধারণ কার্ল কার্যকর করা যেতে পারে:

 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-এর দক্ষতা এবং মাপযোগ্যতায় অবদান রাখে। অ্যাসিনসিওর সাথে একত্রিত হলে, ওআরএম বৈশিষ্ট্যগুলি এপিআই-এ ব্যতিক্রমীভাবে ভাল কাজ করে।

বাকি দুটি শেষ পয়েন্ট কম জটিল এবং মিল রয়েছে। একটি অংশ যা গভীর ব্যাখ্যার দাবি রাখে তা হল get_ingredients ফাংশনের ভিতরে scalars পদ্ধতির ব্যবহার। SQLAlchemy ব্যবহার করে ডাটাবেস অনুসন্ধান করার সময়, execute পদ্ধতিটি প্রায়শই যুক্তি হিসাবে একটি প্রশ্নের সাথে ব্যবহার করা হয়। execute মেথড সারি-এর মতো টিপল ফেরত দিলে, scalars সরাসরি ওআরএম এন্টিটি ফেরত দেয়, এন্ডপয়েন্টকে ক্লিনার করে।

এখন, একই ফাইলে, potions API বিবেচনা করুন:

 @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" যুক্তির জন্য ধন্যবাদ।

APIগুলি কার্যকরী, কিন্তু কোডের সাথে একটি বড় সমস্যা রয়েছে। যদিও SQLAlchemy আপনাকে আপনার ইচ্ছামত ডাটাবেসের সাথে ইন্টারঅ্যাক্ট করার স্বাধীনতা দেয়, এটি Django's Model.objects এর মতো কোনো উচ্চ-স্তরের "ম্যানেজার" ইউটিলিটি অফার করে না। ফলস্বরূপ, আপনাকে এটি নিজেই তৈরি করতে হবে, যা মূলত উপাদান এবং পোশন APIগুলিতে ব্যবহৃত যুক্তি। যাইহোক, যদি আপনি এই যুক্তিটিকে একটি পৃথক স্থানে নিষ্কাশন না করে সরাসরি এন্ডপয়েন্টে রাখেন, তাহলে আপনি প্রচুর ডুপ্লিকেট কোড পাবেন। অতিরিক্তভাবে, প্রশ্ন বা মডেলে পরিবর্তন করা পরিচালনা করা ক্রমশ কঠিন হয়ে উঠবে।

আসন্ন অধ্যায় রিপোজিটরি প্যাটার্ন উপস্থাপন করে: ORM কোড বের করার জন্য একটি মার্জিত সমাধান।

ভান্ডার

রিপোজিটরি প্যাটার্ন ডাটাবেসের সাথে কাজ করার বিবরণকে বিমূর্ত করতে দেয়। 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 পাইথন জেনেরিক ব্যবহার করে, 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 এর ব্যবহার, FastAPI নির্ভরতার সাথে কাজ করার একটি নতুন উপায়। নির্ভরতার রিটার্নের ধরনটিকে DatabaseRepository[db_models.Ingredient] হিসাবে উল্লেখ করে এবং Depends(get_repository(db_models.Ingredient)) এর সাথে এর ব্যবহার ঘোষণা করে আপনি এন্ডপয়েন্ট: repository: IngredientRepository এ সাধারণ টাইপ টীকা ব্যবহার করে শেষ করতে পারেন।

সংগ্রহস্থলের জন্য ধন্যবাদ, শেষ পয়েন্টগুলিকে সমস্ত ORM-সম্পর্কিত বোঝা সঞ্চয় করতে হবে না। এমনকি আরও জটিল ওষুধের ক্ষেত্রে, আপনাকে যা করতে হবে তা হল একই সময়ে দুটি সংগ্রহস্থল ব্যবহার করা।

আপনি ভাবতে পারেন যে দুটি রিপোজিটরি শুরু করলে সেশন দুবার আরম্ভ হবে কিনা। উত্তর হল না। ফাস্টএপিআই নির্ভরতা সিস্টেম একটি একক অনুরোধে একই নির্ভরতা কল ক্যাশ করে। এর মানে হল সেশন ইনিশিয়ালাইজেশন ক্যাশে করা হয় এবং উভয় রিপোজিটরিই একই সেশন অবজেক্ট ব্যবহার করে। SQLAlchemy এবং FastAPI এর সংমিশ্রণের আরেকটি দুর্দান্ত বৈশিষ্ট্য।

API সম্পূর্ণরূপে কার্যকরী এবং একটি পুনঃব্যবহারযোগ্য, উচ্চ-পারফর্মিং ডেটা-অ্যাক্সেস স্তর রয়েছে। পরবর্তী ধাপ হল কিছু এন্ড-টু-এন্ড পরীক্ষা লিখে প্রয়োজনীয়তা পূরণ হয়েছে তা নিশ্চিত করা।

পরীক্ষামূলক

সফ্টওয়্যার বিকাশে পরীক্ষাগুলি গুরুত্বপূর্ণ ভূমিকা পালন করে। প্রকল্পগুলিতে ইউনিট, ইন্টিগ্রেশন এবং এন্ড-টু-এন্ড (E2E) পরীক্ষা থাকতে পারে। যদিও এটি সাধারণত উচ্চ সংখ্যক অর্থপূর্ণ ইউনিট পরীক্ষা করা সর্বোত্তম, তবে সম্পূর্ণ ওয়ার্কফ্লো সঠিকভাবে কাজ করছে তা নিশ্চিত করতে কমপক্ষে কয়েকটি E2E পরীক্ষা লেখাও ভাল।

অ্যালকেমিস্ট অ্যাপের জন্য কিছু E2E পরীক্ষা তৈরি করতে, দুটি অতিরিক্ত লাইব্রেরি প্রয়োজন:

• pytest আসলে তৈরি এবং পরীক্ষা চালানোর জন্য
• httpx পরীক্ষার ভিতরে async অনুরোধ করতে

একবার এগুলি ইনস্টল হয়ে গেলে, পরবর্তী ধাপে একটি পৃথক, পরীক্ষা ডাটাবেস থাকা। আপনি আপনার ডিফল্ট ডাটাবেস দূষিত বা বাদ দিতে চান না। যেহেতু আলকেমিস্ট একটি ডকার সেটআপ অন্তর্ভুক্ত করে, তাই দ্বিতীয় ডাটাবেস তৈরি করার জন্য শুধুমাত্র একটি সাধারণ স্ক্রিপ্ট প্রয়োজন। scripts/create_test_db.sh ফাইল থেকে কোডটি দেখুন:

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

স্ক্রিপ্টটি কার্যকর করার জন্য, এটি পোস্টগ্রেস কন্টেইনারে একটি ভলিউম হিসাবে যোগ করতে হবে। docker-compose.yaml ফাইলের volumes বিভাগে এটি অন্তর্ভুক্ত করে এটি অর্জন করা যেতে পারে।

প্রস্তুতির চূড়ান্ত ধাপ হল tests/conftest.py ফাইলের মধ্যে pytest ফিক্সচার তৈরি করা:

 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

একটি জিনিস যা পরীক্ষায় পরিবর্তন করা অপরিহার্য, তা হল অ্যাপটি কীভাবে ডাটাবেসের সাথে ইন্টারঅ্যাক্ট করে। এর মধ্যে শুধুমাত্র ডাটাবেসের URL পরিবর্তন করাই নয় বরং একটি খালি ডাটাবেস দিয়ে শুরু করে প্রতিটি পরীক্ষা আলাদা করা হয়েছে তা নিশ্চিত করাও অন্তর্ভুক্ত।

db_session ফিক্সচার এই দুটি লক্ষ্যই পূরণ করে। এর শরীর নিম্নলিখিত পদক্ষেপগুলি গ্রহণ করে:

1. একটি পরিবর্তিত ডাটাবেস URL দিয়ে একটি ইঞ্জিন তৈরি করুন।
2. পরীক্ষার একটি পরিষ্কার ডাটাবেস আছে তা নিশ্চিত করতে বিদ্যমান সমস্ত টেবিল মুছুন।
3. ডাটাবেসের ভিতরে সমস্ত টেবিল তৈরি করুন (মাইগ্রেশন স্ক্রিপ্টের মতো একই কোড)।
4. একটি সেশন অবজেক্ট তৈরি করুন এবং ফলন করুন।
5. পরীক্ষা সম্পন্ন হলে ম্যানুয়ালি সেশন বন্ধ করুন।

যদিও শেষ ধাপটি একটি প্রসঙ্গ পরিচালক হিসাবেও প্রয়োগ করা যেতে পারে, ম্যানুয়াল ক্লোজিং এই ক্ষেত্রে ঠিক কাজ করে।

দুটি অবশিষ্ট ফিক্সচার বেশ স্ব-ব্যাখ্যামূলক হওয়া উচিত:

• test_app হল alchemist/app.py ফাইল থেকে FastAPI উদাহরণ, যেখানে 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

পরীক্ষাটি একটি ফিক্সচারে তৈরি একটি ক্লায়েন্ট অবজেক্ট ব্যবহার করে, যা ওভাররাইড নির্ভরতা সহ FastAPI উদাহরণে অনুরোধ করে। ফলস্বরূপ, পরীক্ষাটি একটি পৃথক ডাটাবেসের সাথে ইন্টারঅ্যাক্ট করতে সক্ষম হয় যা পরীক্ষা শেষ হওয়ার পরে পরিষ্কার করা হবে। উভয় API-এর জন্য অবশিষ্ট পরীক্ষা স্যুটের গঠন প্রায় একই।

সারসংক্ষেপ

FastAPI এবং SQLAlchemy আধুনিক এবং শক্তিশালী ব্যাকএন্ড অ্যাপ্লিকেশন তৈরির জন্য চমৎকার প্রযুক্তি। তারা যে স্বাধীনতা, সরলতা এবং নমনীয়তা দেয় তা তাদের পাইথন-ভিত্তিক প্রকল্পগুলির জন্য সেরা বিকল্পগুলির মধ্যে একটি করে তোলে। যদি বিকাশকারীরা সর্বোত্তম অভ্যাস এবং নিদর্শনগুলি অনুসরণ করে, তবে তারা পারফরম্যান্স, শক্তিশালী এবং সুগঠিত অ্যাপ্লিকেশনগুলি তৈরি করতে পারে যা সহজেই ডাটাবেস অপারেশন এবং API লজিক পরিচালনা করে। এই আশ্চর্যজনক সংমিশ্রণটি কীভাবে সেট আপ এবং বজায় রাখতে হয় সে সম্পর্কে আপনাকে ভাল বোঝার জন্য এই নিবন্ধটির লক্ষ্য।

সূত্র

অ্যালকেমিস্ট প্রকল্পের উত্স কোডটি এখানে পাওয়া যাবে: লিঙ্ক ।