Overview
Use cases orchestrate business logic by coordinating repositories and handling duplicate detection. They implement the UseCaseInterface and operate on validated domain models.
CompositeSaveMovieWithActorsUseCase
Orchestrates multiple save operations concurrently using a thread pool.
Class Definition
from concurrent.futures import ThreadPoolExecutor
from typing import List
from domain.models.movie import Movie
from domain.interfaces.use_case_interface import UseCaseInterface
class CompositeSaveMovieWithActorsUseCase(UseCaseInterface):
def __init__(self, use_cases: List[UseCaseInterface]):
self.use_cases = use_cases
self.max_workers = len(use_cases)
application/use_cases/composite_save_movie_with_actors_use_case.py:7-23
Constructor
use_cases
List[UseCaseInterface]
required
List of use cases to execute concurrently (e.g., CSV and PostgreSQL use cases).
Methods
execute
Executes all use cases in parallel using a thread pool.
def execute(self, movie: Movie) -> None
Movie object containing validated data to persist.
application/use_cases/composite_save_movie_with_actors_use_case.py:25-35
Example
from application.use_cases import (
CompositeSaveMovieWithActorsUseCase,
SaveMovieWithActorsCsvUseCase,
SaveMovieWithActorsPostgresUseCase
)
# Create individual use cases
csv_use_case = SaveMovieWithActorsCsvUseCase(
movie_repository=movie_csv_repo,
actor_repository=actor_csv_repo,
movie_actor_repository=relation_csv_repo
)
postgres_use_case = SaveMovieWithActorsPostgresUseCase(
movie_repository=movie_pg_repo,
actor_repository=actor_pg_repo,
movie_actor_repository=relation_pg_repo
)
# Composite use case executes both concurrently
composite = CompositeSaveMovieWithActorsUseCase(
use_cases=[csv_use_case, postgres_use_case]
)
composite.execute(movie) # Saves to both CSV and PostgreSQL in parallel
SaveMovieWithActorsCsvUseCase
Saves a movie and its actors to CSV files, handling duplicates.
Class Definition
from domain.models.movie import Movie
from domain.interfaces.use_case_interface import UseCaseInterface
from domain.repositories.movie_repository import MovieRepository
from domain.repositories.actor_repository import ActorRepository
from domain.repositories.movie_actor_repository import MovieActorRepository
class SaveMovieWithActorsCsvUseCase(UseCaseInterface):
def __init__(
self,
movie_repository: MovieRepository,
actor_repository: ActorRepository,
movie_actor_repository: MovieActorRepository
):
self.movie_repo = movie_repository
self.actor_repo = actor_repository
self.movie_actor_repo = movie_actor_repository
application/use_cases/save_movie_with_actors_csv_use_case.py:12-26
Constructor
Repository for movie persistence.
Repository for actor persistence.
movie_actor_repository
MovieActorRepository
required
Repository for movie-actor relationship persistence.
Methods
execute
Saves movie and actors to CSV, with duplicate detection.
def execute(self, movie: Movie) -> None
Movie object with validated data and associated actors.
application/use_cases/save_movie_with_actors_csv_use_case.py:28-65
Logic Flow
- Check movie duplicates - Search by IMDb ID, skip if exists
- Save movie - Persist to CSV, receive ID
- Process actors - For each actor:
- Check if actor exists by name
- Use existing or save new actor
- Collect movie-actor relationships
- Save relationships - Batch save all relationships
Error Handling
Logs errors and continues execution:
try:
# ... save operations
except Exception as e:
logger.error(f"Error al escribir en CSV para la película '{movie.title}': {e}")
application/use_cases/save_movie_with_actors_csv_use_case.py:64-65
SaveMovieWithActorsPostgresUseCase
Saves a movie and its actors to PostgreSQL database, handling duplicates.
Class Definition
from domain.models.movie import Movie
from domain.interfaces.use_case_interface import UseCaseInterface
from domain.repositories.movie_repository import MovieRepository
from domain.repositories.actor_repository import ActorRepository
from domain.repositories.movie_actor_repository import MovieActorRepository
class SaveMovieWithActorsPostgresUseCase(UseCaseInterface):
def __init__(
self,
movie_repository: MovieRepository,
actor_repository: ActorRepository,
movie_actor_repository: MovieActorRepository
):
self.movie_repo = movie_repository
self.actor_repo = actor_repository
self.movie_actor_repo = movie_actor_repository
application/use_cases/save_movie_with_actors_postgres_use_case.py:15-28
Constructor
Repository for movie persistence.
Repository for actor persistence.
movie_actor_repository
MovieActorRepository
required
Repository for movie-actor relationship persistence.
Methods
execute
Saves movie and actors to PostgreSQL with duplicate handling.
def execute(self, movie: Movie) -> None
Movie object with validated data and associated actors.
application/use_cases/save_movie_with_actors_postgres_use_case.py:30-69
Logic Flow
- Check movie duplicates - Query by IMDb ID, skip if exists
- Save movie - Use stored procedure, verify ID returned
- Process actors - For each actor:
- Query if actor exists by name
- Use existing or insert new actor
- Collect valid movie-actor relationships
- Save relationships - Batch insert relationships
Error Handling
Logs database errors with context:
try:
# ... database operations
except Exception as e:
logger.error(f"Error en la base de datos al procesar '{movie.title}': {e}")
application/use_cases/save_movie_with_actors_postgres_use_case.py:68-69
Additional Validation
Verifies successful save before continuing:
saved_movie = self.movie_repo.save(movie)
if not saved_movie or not saved_movie.id:
logger.error(f"Error al guardar la película '{movie.title}' o al obtener su ID.")
return
application/use_cases/save_movie_with_actors_postgres_use_case.py:45-48
Common Patterns
Duplicate Prevention
All use cases check for existing entities:
# Check if movie already exists
existing_movie = self.movie_repo.find_by_imdb_id(movie.imdb_id)
if existing_movie:
logger.info(f"La película '{movie.title}' ya existe. Saltando.")
return
# Check if actor exists
existing_actor = self.actor_repo.find_by_name(actor.name)
if existing_actor:
saved_actor = existing_actor
else:
saved_actor = self.actor_repo.save(actor)
Batch Operations
Collect relationships and save in bulk:
relations_to_save = []
for actor in movie.actors:
# ... process actor
relations_to_save.append(
MovieActor(movie_id=saved_movie.id, actor_id=saved_actor.id)
)
if relations_to_save:
self.movie_actor_repo.save_many(relations_to_save)
Dependency Injection
Repositories injected via constructor, enabling:
- Flexibility - Swap implementations
- Testability - Use mock repositories
- Separation - Business logic independent of infrastructure
# Production
use_case = SaveMovieWithActorsCsvUseCase(
movie_repository=MovieCsvRepository(),
actor_repository=ActorCsvRepository(),
movie_actor_repository=MovieActorCsvRepository()
)
# Testing
use_case = SaveMovieWithActorsCsvUseCase(
movie_repository=MockMovieRepository(),
actor_repository=MockActorRepository(),
movie_actor_repository=MockRelationRepository()
)