⚡ Section 6 — Mid-Level Masterclass
The Repository Pattern & Data Access Decoupling
Stop leaking raw ORM queries into HTTP routes and business logic. Abstract data access behind domain-driven repository contracts.
👁️ Views: 362 Page Views
💡 Mindset Transformation
Leaked ORM Queries vs Clean Repository Pattern
Instead of spreading raw database queries throughout your application codebase, encapsulate data persistence behind abstract repository methods.
❌ Leaked ORM Queries Scattered in Controllers
# ❌ Tight coupling to ORM and Database driver
@app.route("/users/")
def get_user(user_id):
user = db.query(User).filter(User.id == user_id).first()
if not user:
abort(404)
return jsonify(user.to_dict())
✅ Clean Repository Pattern Interface
# ✅ Clean domain service calling abstract repository
class UserService:
def __init__(self, user_repo: UserRepository) -> None:
self.user_repo = user_repo
def get_user_profile(self, user_id: int) -> User:
user = self.user_repo.get_by_id(user_id)
return user🏛️ Architectural Isolation Layers
Business Logic (UserService)
➔
Repository Interface (UserRepository Contract)
➔
SqlAlchemyUserRepository (Production SQL)
InMemoryUserRepository (Instant Mock Test)
💻 Production Code Blueprint: Abstract Contract & Dual Implementations
Notice how swapping between a real PostgreSQL database adapter and an in-memory dictionary mock requires changing only the repository dependency injection.
# ─── 1. ABSTRACT REPOSITORY CONTRACT ──────────────────────────────────────
from abc import ABC, abstractmethod
from typing import Optional
class UserRepository(ABC):
@abstractmethod
def get_by_id(self, user_id: int) -> Optional[User]:
...
# ─── 2. PRODUCTION SQLALCHEMY IMPLEMENTATION ──────────────────────────────
class SqlAlchemyUserRepository(UserRepository):
def __init__(self, db_session) -> None:
self.session = db_session
def get_by_id(self, user_id: int) -> Optional[User]:
orm_user = self.session.query(UserModel).filter(UserModel.id == user_id).first()
return User(id=orm_user.id, email=orm_user.email) if orm_user else None
# ─── 3. IN-MEMORY MOCK REPOSITORY (FOR LIGHTNING UNIT TESTS) ──────────────
class InMemoryUserRepository(UserRepository):
def __init__(self) -> None:
self._users: dict[int, User] = {}
def get_by_id(self, user_id: int) -> Optional[User]:
return self._users.get(user_id)