Persistence Layer

import csv
import threading
from typing import Optional
from domain.models.movie import Movie
from domain.repositories.movie_repository import MovieRepository

class MovieCsvRepository(MovieRepository):
    def __init__(self):
        os.makedirs(os.path.dirname(MOVIES_CSV), exist_ok=True)
        if not os.path.exists(MOVIES_CSV):
            with open(MOVIES_CSV, "w", newline="", encoding="utf-8") as f:
                writer = csv.writer(f)
                writer.writerow(MOVIE_HEADERS)

def save(self, movie: Movie) -> Movie

def find_by_imdb_id(self, imdb_id: str) -> Optional[Movie]

from domain.models.actor import Actor
from domain.repositories.actor_repository import ActorRepository

class ActorCsvRepository(ActorRepository):
    def __init__(self):
        os.makedirs(os.path.dirname(ACTORS_CSV), exist_ok=True)
        if not os.path.exists(ACTORS_CSV):
            with open(ACTORS_CSV, "w", newline="", encoding="utf-8") as f:
                writer = csv.writer(f)
                writer.writerow(ACTOR_HEADERS)

def save(self, actor: Actor) -> Actor

def find_by_name(self, name: str) -> Optional[Actor]

from domain.models.movie_actor import MovieActor
from domain.repositories.movie_actor_repository import MovieActorRepository

class MovieActorCsvRepository(MovieActorRepository):
    def __init__(self):
        os.makedirs(os.path.dirname(MOVIE_ACTOR_CSV), exist_ok=True)
        if not os.path.exists(MOVIE_ACTOR_CSV):
            with open(MOVIE_ACTOR_CSV, "w", newline="", encoding="utf-8") as f:
                writer = csv.writer(f)
                writer.writerow(MOVIE_ACTOR_HEADERS)

def save(self, relation: MovieActor) -> None

def save_many(self, relations: List[MovieActor]) -> None

from domain.models.movie import Movie
from domain.repositories.movie_repository import MovieRepository
from psycopg2 import DatabaseError

class MoviePostgresRepository(MovieRepository):
    def __init__(self, conn):
        self.conn = conn

def save(self, movie: Movie) -> Movie

SELECT * FROM upsert_movie(
    imdb_id, title, year, rating, duration_minutes, metascore
)

def find_by_imdb_id(self, imdb_id: str) -> Optional[Movie]

SELECT id, imdb_id, title, year, rating, duration_minutes, metascore 
FROM movies 
WHERE imdb_id = %s

from domain.models.actor import Actor
from domain.repositories.actor_repository import ActorRepository

class ActorPostgresRepository(ActorRepository):
    def __init__(self, conn):
        self.conn = conn

def save(self, actor: Actor) -> Actor

SELECT * FROM upsert_actor(name)

def find_by_name(self, name: str) -> Optional[Actor]

from domain.models.movie_actor import MovieActor
from domain.repositories.movie_actor_repository import MovieActorRepository

class MovieActorPostgresRepository(MovieActorRepository):
    def __init__(self, conn):
        self.conn = conn

def save(self, relation: MovieActor) -> None

SELECT * FROM upsert_movie_actor(movie_id, actor_id)

def save_many(self, relations: List[MovieActor]) -> None

movie_lock = threading.Lock()

def save(self, movie: Movie) -> Movie:
    with movie_lock:
        # ... atomic file operations

from psycopg2.pool import ThreadedConnectionPool

pool = ThreadedConnectionPool(
    minconn=1,
    maxconn=10,
    host="localhost",
    database="imdb_scraper",
    user="user",
    password="password"
)

conn = pool.getconn()
repo = MoviePostgresRepository(conn)

try:
    saved_movie = movie_repo.save(movie)
except IOError as e:
    logger.error(f"File write error: {e}")
except Exception as e:
    logger.error(f"Unexpected error: {e}")

from psycopg2 import DatabaseError

try:
    saved_movie = movie_repo.save(movie)
except DatabaseError as e:
    logger.error(f"Database error: {e}")
    conn.rollback()
    raise

from infrastructure.persistence.csv.repositories import (
    MovieCsvRepository,
    ActorCsvRepository,
    MovieActorCsvRepository
)
from domain.models.movie import Movie
from domain.models.actor import Actor
from domain.models.movie_actor import MovieActor

# Initialize repositories
movie_repo = MovieCsvRepository()
actor_repo = ActorCsvRepository()
relation_repo = MovieActorCsvRepository()

# Save 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)
print(f"Saved movie with ID: {saved_movie.id}")

# Save actors
actor = Actor(id=None, name="Tim Robbins")
saved_actor = actor_repo.save(actor)

# Save relationship
relation = MovieActor(
    movie_id=saved_movie.id,
    actor_id=saved_actor.id
)
relation_repo.save(relation)

import psycopg2
from infrastructure.persistence.postgres.repositories import (
    MoviePostgresRepository,
    ActorPostgresRepository,
    MovieActorPostgresRepository
)

# Connect to database
conn = psycopg2.connect(
    host="localhost",
    database="imdb_scraper",
    user="user",
    password="password"
)

# Initialize repositories
movie_repo = MoviePostgresRepository(conn)
actor_repo = ActorPostgresRepository(conn)
relation_repo = MovieActorPostgresRepository(conn)

# Save movie (same API as CSV)
saved_movie = movie_repo.save(movie)
conn.commit()

# Close connection
conn.close()