Skip to main content

​
ADR-007: Repository Pattern with SQLModel Integration

​
Context

The framework needed standardized data persistence patterns. Ad-hoc database access led to inconsistent implementations, tight coupling between business logic and database details, complex session management, and difficult testing.

​
Decision

Implemented a Repository Pattern with SQLModel integration providing standardized data access through BaseRepository and SQLModelRepository classes.

​
Key Components

  • BaseRepository: Abstract base class defining repository interface
  • SQLModelRepository: Concrete SQLModel implementation with session management
  • Context Manager Pattern: Automatic session handling with rollback on exceptions
  • Dependency Injection: Seamless integration with flavor system
  • Configuration-Driven: Database setup through config classes

​
Core Implementation

class SQLModelRepository(BaseRepository):
    def __init__(self, database_url: Optional[str] = None, echo: bool = False):
        super().__init__(database_url, echo)
        if database_url:
            self._engine = create_engine(self.database_url, echo=self.echo)
    
    @contextmanager
    def get_session(self):
        with Session(self._engine) as session:
            try:
                yield session
            except Exception:
                session.rollback()
                raise

​
Consequences

​
Positive

  • Consistent data access patterns across flavors
  • Improved testability through repository abstraction
  • Automatic session management prevents resource leaks
  • Clear separation between data access and business logic
  • Type safety with SQLModel integration
  • Easy mocking for unit tests

​
Negative

  • Additional abstraction layer complexity
  • Learning curve for repository pattern concepts
  • Minimal performance overhead from context managers
  • Framework dependency on SQLModel

​
Usage Example

class UserRepository(SQLModelRepository):
    def __init__(self, config: MyConfig):
        super().__init__(database_url=config.database_url)
    
    def get_user_by_id(self, user_id: int) -> Optional[User]:
        with self.get_session() as session:
            return session.get(User, user_id)

class UserService(BaseService):
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository
    
    def get_user(self, user_id: int):
        return self.user_repository.get_user_by_id(user_id)

class MyFlavor(BaseFlavor):
    def __init__(self):
        super().__init__(
            services=[UserService],
            repositories=[UserRepository],
            config=MyConfig
        )
Session management is automatic with proper rollback on exceptions.
⌘I