This article was originally published on AI Study Room. For the full version with working code examples and related articles, visit the original post.
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Database Design Patterns: Repository, Unit of Work, Query Objects, Table Inheritance
Design patterns provide reusable solutions to common database access problems. This article covers patterns that help decouple business logic from database access, manage transactions, and model inheritance.
Repository Pattern
The Repository pattern mediates between the domain model and the database, providing a collection-like interface for accessing domain objects.
Interface
from abc import ABC, abstractmethod
from typing import Optional, List
class UserRepository(ABC):
def find_by_id(self, user_id: int) -> Optional[dict]:
pass
def find_by_email(self, email: str) -> Optional[dict]:
pass
def save(self, user: dict) -> int:
pass
def delete(self, user_id: int) -> bool:
pass
Implementation
import psycopg2
from psycopg2.extras import RealDictCursor
class PostgresUserRepository(UserRepository):
def init(self, conn):
self.conn = conn
def find_by_id(self, user_id: int) -> Optional[dict]:
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT * FROM users WHERE id = %s",
(user_id,)
)
return cur.fetchone()
def find_by_email(self, email: str) -> Optional[dict]:
with self.conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(
"SELECT * FROM users WHERE email = %s",
(email,)
)
return cur.fetchone()
def save(self, user: dict) -> int:
with self.conn.cursor() as cur:
if 'id' in user:
cur.execute("""
UPDATE users SET email = %s, name = %s
WHERE id = %s RETURNING id
""", (user['email'], user['name'], user['id']))
else:
cur.execute("""
INSERT INTO users (email, name)
VALUES (%s, %s) RETURNING id
""", (user['email'], user['name']))
return cur.fetchone()[0]
def delete(self, user_id: int) -> bool:
with self.conn.cursor() as cur:
cur.execute(
"DELETE FROM users WHERE id = %s", (user_id,)
)
return cur.rowcount > 0
Benefits
Testability : Mock the repository interface for unit tests.
Encapsulation : SQL is contained within the repository.
Swappable : Change database implementation without changing business logic.
Query optimization : Changes are isolated to the repository.
Unit of Work Pattern
Unit of Work tracks changes to objects during a business transaction and writes them as a single unit:
class UnitOfWork:
def init(self, conn):
self.conn = conn
self.new_objects = []
self.dirty_objects = []
self.deleted_objects = []
self.repositories = {}
def register_new(self, obj):
self.new_objects.append(obj)
def register_dirty(self, obj):
if obj not in self.dirty_objects:
self.dirty_objects.append(obj)
def register_deleted(self, obj):
self.deleted_objects.append(obj)
def commit(self):
if not self.new_objects and not self.dirty_objects and not self.deleted_objects:
return
with self.conn:
for obj in self.deleted_objects:
self._delete(obj)
for obj in self.dirty_objects:
self._update(obj)
for obj in self.new_objects:
self._insert(obj)
self.new_objects.clear()
self.dirty_objects.clear()
self.deleted_objects.clear()
def _insert(self, obj):
repo = self._get_repo(type(obj))
repo.save(obj)
def _update(self, obj):
repo = self._get_repo(type(obj))
repo.save(obj)
def _delete(self, obj):
repo = self._get_repo(type(obj))
repo.delete(obj.id)
def _get_repo(self, obj_type):
Repository registry determines which repository maps to which type
pass
Usage
def create_order(user_id, items):
uow = UnitOfWork(connection)
user_repo = UserRepository(uow)
order_repo = OrderRepository(uow)
user = user_repo.find_by_id(user_id)
user['last_order_date'] = datetime.utcnow()
uow.register_dirty(user)
order = {'user_id': user_id, 'items': items, 'total': calculate_total(items)}
uow.register_new(
Read the full article on AI Study Room for complete code examples, comparison tables, and related resources.
Found this useful? Check out more developer guides and tool comparisons on AI Study Room.
Top comments (0)