Overview
Repository interfaces define the contract for data persistence operations. Concrete implementations can use CSV, PostgreSQL, or other storage mechanisms.
MovieRepository
Interface for movie persistence operations.
Class Definition
from abc import ABC, abstractmethod
from typing import Optional
from domain.models.movie import Movie
class MovieRepository(ABC):
@abstractmethod
def save(self, movie: Movie) -> Movie:
pass
@abstractmethod
def find_by_imdb_id(self, imdb_id: str) -> Optional[Movie]:
pass
domain/repositories/movie_repository.py:5-38
Methods
save
Persists a movie and returns the saved entity with any generated fields.
def save(self, movie: Movie) -> Movie
Movie object containing the information to save.
The saved Movie entity, including any database-generated fields (e.g., id).
domain/repositories/movie_repository.py:13-25
find_by_imdb_id
Searches for a movie by its IMDb ID to prevent duplicates.
def find_by_imdb_id(self, imdb_id: str) -> Optional[Movie]
IMDb unique identifier (e.g., tt0111161).
The Movie object if found, otherwise None.
domain/repositories/movie_repository.py:27-38
Implementations
ActorRepository
Interface for actor persistence operations.
Class Definition
from abc import ABC, abstractmethod
from typing import Optional
from domain.models.actor import Actor
class ActorRepository(ABC):
@abstractmethod
def save(self, actor: Actor) -> Actor:
pass
@abstractmethod
def find_by_name(self, name: str) -> Optional[Actor]:
pass
domain/repositories/actor_repository.py:5-28
Methods
save
Persists an actor and returns the saved entity.
def save(self, actor: Actor) -> Actor
The saved Actor entity with database-generated ID.
domain/repositories/actor_repository.py:10-15
find_by_name
Searches for an actor by name to prevent duplicates.
def find_by_name(self, name: str) -> Optional[Actor]
Actor’s full name to search for.
The Actor object if found, otherwise None.
domain/repositories/actor_repository.py:17-28
Implementations
MovieActorRepository
Interface for managing movie-actor relationships (N:M).
Class Definition
from abc import ABC, abstractmethod
from typing import List
from domain.models.movie_actor import MovieActor
class MovieActorRepository(ABC):
@abstractmethod
def save(self, relation: MovieActor) -> None:
pass
@abstractmethod
def save_many(self, relations: List[MovieActor]) -> None:
pass
domain/repositories/movie_actor_repository.py:5-33
Methods
save
Persists a single movie-actor relationship.
def save(self, relation: MovieActor) -> None
The relationship to persist.
domain/repositories/movie_actor_repository.py:10-20
Useful for adding a single association. For bulk operations, use save_many for better performance.
save_many
Persists multiple movie-actor relationships efficiently.
def save_many(self, relations: List[MovieActor]) -> None
List of MovieActor relationships to persist.
domain/repositories/movie_actor_repository.py:22-33
Preferred method when processing a movie with multiple actors, as it optimizes write operations.
Implementations
Usage Example
from domain.models.movie import Movie
from domain.models.actor import Actor
from domain.models.movie_actor import MovieActor
from infrastructure.persistence.csv.repositories import (
MovieCsvRepository,
ActorCsvRepository,
MovieActorCsvRepository
)
# Initialize repositories
movie_repo = MovieCsvRepository()
actor_repo = ActorCsvRepository()
relation_repo = MovieActorCsvRepository()
# Save a movie
movie = Movie(
id=None,
imdb_id="tt0111161",
title="The Shawshank Redemption",
year=1994,
rating=9.3,
duration_minutes=142,
metascore=82,
actors=[]
)
saved_movie = movie_repo.save(movie)
# Save actors and relationships
for actor_name in ["Tim Robbins", "Morgan Freeman"]:
# Check if actor exists
existing_actor = actor_repo.find_by_name(actor_name)
if existing_actor:
saved_actor = existing_actor
else:
actor = Actor(id=None, name=actor_name)
saved_actor = actor_repo.save(actor)
# Create relationship
relation = MovieActor(
movie_id=saved_movie.id,
actor_id=saved_actor.id
)
relation_repo.save(relation)
Repository Pattern Benefits
Abstraction
Business logic doesn’t depend on storage mechanism:
class SaveMovieUseCase:
def __init__(self, movie_repository: MovieRepository):
self.repo = movie_repository # Works with any implementation
Testing
Easy to create in-memory implementations for tests:
class InMemoryMovieRepository(MovieRepository):
def __init__(self):
self.movies = {}
def save(self, movie: Movie) -> Movie:
movie.id = len(self.movies) + 1
self.movies[movie.imdb_id] = movie
return movie
def find_by_imdb_id(self, imdb_id: str) -> Optional[Movie]:
return self.movies.get(imdb_id)
Flexibility
Swap storage implementations without changing business logic:
# Development: Use CSV
repo = MovieCsvRepository()
# Production: Use PostgreSQL
repo = MoviePostgresRepository(connection)
# Testing: Use in-memory
repo = InMemoryMovieRepository()