Patrón Repository en Python

Patrón Repository en Python

Separar todas las funcionalidades relacionadas con el manejo de datos en nuestras aplicaciones puede resultar una buena idea, especialmente cuando desarrollamos operaciones más allá de un simple CRUD. En esta entrada, te voy a explicar el patrón repositorio (Repository Pattern), el cual nos ofrece una forma de separar en una capa dedicada todo lo relacionado con el manejo de datos del resto de nuestra aplicación.

¿Qué es el patrón repositorio?

Entrando en más detalle, este patrón nos ofrece una capa de abstracción mediante el uso de interfaces o clases abstractas que esconden todas las operaciones relacionadas con la infraestructura de datos. De esta forma, el dominio de nuestro software queda aislado y no se preocupa de cómo se cargan o se obtienen los datos; eso se lo dejamos a nuestro repositorio.

En varios lugares ponen el ejemplo de que con un repositorio se te puede facilitar cambiar de motor o tipo de base de datos, porque precisamente todas las consultas y procesos para el manejo de los datos están aislados en el repositorio. Pero seamos sinceros, es muy raro cambiar de base de datos en un proyecto. Aquí te dejo algunas ventajas que creo que pueden hacer que tener un repositorio te haga la vida más fácil:

  • Testing: Sin duda, una de las mayores ventajas de un repositorio. Con este patrón implementado, hacer pruebas en nuestra aplicación puede ser tan sencillo como crear un mock y despreocuparnos de cómo se hacen las cosas tras bambalinas con el manejo de datos.

  • Mejora la Mantenibilidad: Al separar la lógica de acceso a datos del resto del código, se facilita el mantenimiento y la evolución del software. Los cambios en la lógica de acceso a datos se pueden realizar en un solo lugar.

  • Fomenta el Principio de Responsabilidad Única: Cada repositorio se encarga de una entidad específica, promoviendo el principio de responsabilidad única (SRP) y haciendo que el código sea más modular y fácil de entender.

    Facilita la Implementación de Patrones Avanzados: Implementar patrones avanzados como Unit of Work o CQRS (Command Query Responsibility Segregation) se vuelve más sencillo cuando se utiliza el patrón repositorio.

  • Consistencia y Gestión de Transacciones: Los repositorios pueden gestionar transacciones y asegurar la consistencia de los datos, lo que es esencial en aplicaciones que requieren integridad de datos.

¿Cómo se implementa este patrón?

Como buen patrón de diseño, este ya nos trae reglas específicas y un camino propuesto a seguir, ¿obviamente no? Si no, no sería un patrón.

UML Repository

 

Primero, vamos a definir una implementación muy sencilla solo para ejemplificar las interfaces y clases mínimas que necesitamos. Luego, veremos un caso de uso un poco más aplicado a algo real.

Siguiendo el diagrama de implementación, lo primero que tenemos que hacer es definir nuestra interfaz genérica de un repositorio. (En esta entrada ya hablamos de cómo implementar clases abstractas para cubrir el propósito de las interfaces en Python por si quieres echarle un vistazo.) En este caso, cuando me refiera a interfaz, verás una clase abstracta.

from typing import Generic, Optional, TypeVar
from abc import ABC, abstractmethod

T = TypeVar('T')


class IRepository(ABC, Generic[T]):\r\n    
   """Interfaz generica para los repositorios"""
   
   @abstractmethod
   def get_by_id(self, id: int) -> Optional[T]:
      raise NotImplementedError
    
   @abstractmethod
   def create(self, item: T) -> None:
      raise NotImplementedError

La interfaz IRepository es la que define los requerimientos mínimos que debe tener nuestro repositorio. En este caso, solo incluye operaciones simples como get_by_id() y create().

Ahora podemos crear repositorios concretos para las distintas entidades de nuestro proyecto. Vamos a definir una clase muy simple que será la entidad con la que vamos a trabajar, en este caso un simple post.

class Post:
    """
    Modelo de Post
    """

    def __init__(self, id: int, title: str, content: str):
        self.id = id
        self.title = title
        self.content = content

    def __repr__(self):
        return f"Post(id={self.id}, title={self.title}, content={self.content})"

Ya que tenemos la entidad para la que vamos a definir nuestro repositorio, vamos a definir un repositorio concreto para manejar los datos de los posts, en este caso un simple repositorio en memoria.

from typing import Optional, List

class InMemoryPostRepository(IRepository[Post]):
    """
    Repositorio de Post en memoria
    """

    def __init__(self):
        self._data_source: List[Post] = []

    def get_by_id(self, id: int) -> Optional[Post]:
        for item in self._data_source:
            if hasattr(item, 'id') and item.id == id:
                return item
        return None

    def create(self, item: Post) -> None:
        self._data_source.append(item)

Listo, este sería nuestro sencillo repositorio. Como puedes ver, implementa la interfaz genérica del repositorio, pasando como referencia el tipo de objetos con los que va a trabajar.

Hasta este punto, ya hemos terminado con la implementación del repositorio; lo que nos dice el diagrama UML ya está hecho.

Lo que sigue es una práctica muy común: definir un servicio que será el encargado de interactuar con nuestro repositorio. Es decir, este será el intermediario entre nuestro repositorio y el resto de la aplicación.

from typing import Optional

class PostService:
    """
    Servicio de Post
    """

    def __init__(self, repository: InMemoryPostRepository):
        self._repository = repository

    def get_post_by_id(self, id: int) -> Optional[Post]:
        return self._repository.get_by_id(id)

    def create_post(self, id: int, title: str, content: str) -> None:
        post = Post(id, title, content)
        self._repository.create(post)

El servicio simplemente tiene un mapeo de los métodos que tenemos disponibles en el repositorio. Este recibe como dependencia el repositorio con el que va a trabajar.

Ahora sí, podemos probar el funcionamiento de nuestro repositorio:

# Creamos un repositorio de Post en memoria
post_repo = InMemoryPostRepository()

# Lo inyectamos en el servicio de Post
post_service = PostService(post_repo)

# Listo, ahora podemos crear y obtener posts
post_service.create_post(1, "Post1", "Content1")
post = post_service.get_post_by_id(1)

print(post)
# >>> Post(id=1, title=Post1, content=Content1)

Listo, ya tenemos nuestro patrón repositorio funcionando. Como te podrás dar cuenta, el proceso del manejo de la información se vuelve muy simple: nosotros solo llamamos a los métodos de nuestro servicio, este hace la comunicación con el repositorio, y luego el repositorio se encarga de ejecutar y manejar los procesos del manejo de datos.

Caso de uso un poco mas completo

Ahora que ya tenemos una idea de cómo se implementa y cómo funciona, quiero mostrar una implementación con un poco más de cosas y cómo podría ser una forma de estructurarlo en algún proyecto.

Vamos a usar como ejemplo el caso de una simple aplicación para mostrar películas, y que un usuario pueda hacer la review de cualquier película. Entonces vamos a proponer la siguiente estructura de proyecto:

my_super_movies_app/
│
├── repository/
│   ├── __init__.py
│   ├── repository_generic.py
│   ├── user_repository.py
│   ├── movie_repository.py
│   └── review_repository.py
│
├── models/
│   ├── __init__.py
│   ├── user.py
│   ├── movie.py
│   └── review.py
│
├── services/
│   ├── __init__.py
│   ├── user_service.py
│   ├── movie_service.py
│   └── review_service.py
│
├── main.py
└── requirements.txt

Básicamente separamos las funcionalidades: un directorio para los repositorios, uno para los modelos y otro para los servicios. Cabe destacar que en este caso esto puede ser fácilmente mantenible. Dependiendo de las entidades que tenga el proyecto, la estructura del proyecto puede cambiar para ser más manejable.

Primero vamos a definir nuestros modelos. Los modelos son básicamente una estructura de cómo luce una entidad de información en una aplicación. Los podemos definir con clases o interfaces, pero normalmente se hacen mediante clases.

¿Cómo va a lucir un usuario en el flujo de datos de nuestro programa? Vamos a crear su modelo.

class User:
    """
    User model
    """

    def __init__(self, username: str, email: str, password: str, date_joined: str = None, id: int = None):
        self.id = id
        self.username = username
        self.email = email
        self.password = password
        self.date_joined = date_joined

De manera similar, podemos definir el modelo para una película y para una review. Aquí te muestro cómo podrían lucir:

class Review:
    """
    Review model
    """

    def __init__(
        self,
        user_id: int,
        movie_id: int,
        review_text: str,
        rating: float,
        review_date: str = None,
        id: int = None
    ):
        self.id = id
        self.movie_id = movie_id
        self.user_id = user_id
        self.review_text = review_text
        self.rating = rating
        self.review_date = review_date

El siguiente paso, como ya vimos, es comenzar a crear la interfaz genérica para el repositorio en el archivo repository_generic.py.

from abc import ABC, abstractmethod
from typing import TypeVar, Generic, List, Optional

T = TypeVar('T')

class IRepository(ABC, Generic[T]):
    """
    Interface for generic repository
    """

    @abstractmethod
    def get_all(self) -> List[T]:
        raise NotImplementedError

    @abstractmethod
    def get_by_id(self, id: int) -> Optional[T]:
        raise NotImplementedError

    @abstractmethod
    def create(self, item: T) -> None:
        raise NotImplementedError

    @abstractmethod
    def update(self, item: T) -> None:
        raise NotImplementedError

    @abstractmethod
    def delete(self, id: int) -> None:
        raise NotImplementedError

Ahora, definamos más métodos CRUD para que cada entidad tenga como base estas funciones.

A continuación, debemos crear los repositorios para nuestras tres entidades: usuarios, películas y reseñas. En estos repositorios concretos es donde, como recordamos, se encuentra la lógica para el manejo de los datos. Por lo tanto, aquí es donde radica el motor de base de datos que utilizaremos. Para este caso, emplearé una base de datos SQLite, ya que es adecuada solo para demostración.

Comencemos con el repositorio para usuarios.

from sqlite3 import Connection
from typing import List, Optional

from PatronRepository.models.user import User
from PatronRepository.repository.repository_generic import IRepository

class UserRepository(IRepository[User]):
    def __init__(self, db_connection: Connection):
        self._connection = db_connection

    def get_all(self) -> List[User]:
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM users")
        results = cursor.fetchall()
        return [User(*result) for result in results]  # Assuming User constructor matches the columns

    def get_by_id(self, id: int) -> Optional[User]:
        cursor = self._connection.cursor()
        cursor.execute('SELECT * FROM users WHERE id=?', (id,))
        result = cursor.fetchone()
        return User(*result) if result is not None else None  # Assuming User constructor matches the columns

    def create(self, item: User) -> User:
        cursor = self._connection.cursor()
        cursor.execute("""
            INSERT INTO users(username, email, password) VALUES (?, ?, ?)
        """, (item.username, item.email, item.password))
        self._connection.commit()
        item.id = cursor.lastrowid
        return item

    def update(self, item: User) -> None:
        cursor = self._connection.cursor()
        cursor.execute("""
            UPDATE users SET username=?, email=?, password=? WHERE id=?
        """, (item.username, item.email, item.password, item.id))
        self._connection.commit()

    def delete(self, id: int) -> None:
        cursor = self._connection.cursor()
        cursor.execute("DELETE FROM users WHERE id=?", (id,))
        self._connection.commit()

El repositorio concreto de usuarios implementa la interfaz IRepository para un usuario. En la implementación de los métodos, ya tenemos las consultas necesarias para insertar, actualizar, eliminar y leer información de nuestra base de datos SQLite.

Continuemos con el repositorio concreto para películas:

from sqlite3 import Connection
from typing import List, Optional

from PatronRepository.models.movie import Movie
from PatronRepository.repository.repository_generic import IRepository

class MoviesRepository(IRepository[Movie]):
    def __init__(self, db_connection: Connection):
        self._connection = db_connection

    def get_all(self) -> List[Movie]:
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM movies")
        results = cursor.fetchall()
        return [Movie(*result) for result in results]  # Assuming Movie constructor matches the columns

    def get_by_id(self, id: int) -> Optional[Movie]:
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM movies WHERE id=?", (id,))
        result = cursor.fetchone()
        return Movie(*result) if result is not None else None  # Assuming Movie constructor matches the columns

    def create(self, item: Movie) -> Movie:
        cursor = self._connection.cursor()
        cursor.execute("""
            INSERT INTO movies(title, genre, director, release_year, rating) VALUES (?, ?, ?, ?, ?)
        """, (item.title, item.genre, item.director, item.release_year, item.rating))
        self._connection.commit()
        item.id = cursor.lastrowid
        return item

    def update(self, item: Movie) -> bool:
        cursor = self._connection.cursor()
        cursor.execute("""
            UPDATE movies SET title=?, genre=?, director=?, release_year=?, rating=? WHERE id=?
        """, (item.title, item.genre, item.director, item.release_year, item.rating, item.id))
        self._connection.commit()
        return cursor.rowcount > 0

    def delete(self, id: int) -> bool:
        cursor = self._connection.cursor()
        cursor.execute("DELETE FROM movies WHERE id=?", (id,))
        self._connection.commit()
        return cursor.rowcount > 0

Y para finalizar con los repositorios, el de reseñas:

from sqlite3 import Connection
from typing import List, Optional

from PatronRepository.models.review import Review
from PatronRepository.repository.repository_generic import IRepository

class ReviewRepository(IRepository[Review]):
    def __init__(self, db_connection: Connection):
        self._connection = db_connection

    def get_all(self) -> List[Review]:
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM reviews")
        results = cursor.fetchall()
        return [Review(*result) for result in results]  # Assuming Review constructor matches the columns

    def get_by_id(self, id: int) -> Optional[Review]:
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM reviews WHERE id=?", (id,))
        result = cursor.fetchone()
        return Review(*result) if result is not None else None  # Assuming Review constructor matches the columns

    def create(self, item: Review) -> None:
        cursor = self._connection.cursor()
        cursor.execute("""
            INSERT INTO reviews(movie_id, user_id, rating, review_text) VALUES (?, ?, ?, ?)
        """, (item.movie_id, item.user_id, item.rating, item.review_text))
        self._connection.commit()

    def update(self, item: Review) -> None:
        cursor = self._connection.cursor()
        cursor.execute("""
            UPDATE reviews SET movie_id=?, user_id=?, rating=?, review_text=? WHERE id=?
        """, (item.movie_id, item.user_id, item.rating, item.review_text, item.id))
        self._connection.commit()

    def delete(self, id: int) -> None:
        cursor = self._connection.cursor()
        cursor.execute("DELETE FROM reviews WHERE id=?", (id,))
        self._connection.commit()

Ahora que tenemos nuestros tres repositorios concretos para cada entidad, el siguiente paso es crear nuestros servicios para interactuar con los repositorios. Como ya vimos, el servicio necesitará que le inyectemos como dependencia el repositorio concreto con el que va a trabajar.

from PatronRepository.models.user import User
from PatronRepository.repository.user_repository import UserRepository

class UserService:
    """
    User Service
    """

    def __init__(self, repository: UserRepository):
        self._repository = repository

    def get_all_users(self) -> List[User]:
        return self._repository.get_all()

    def get_user_by_id(self, user_id: int) -> Optional[User]:
        return self._repository.get_by_id(user_id)

    def create_user(self, username: str, email: str, password: str) -> User:
        new_user = User(username=username, email=email, password=password)
        return self._repository.create(new_user)

    def update_user(self, user_id: int, username: str, email: str, password: str) -> bool:
        user = self._repository.get_by_id(user_id)
        if user is None:
            return False
        user.username = username
        user.email = email
        user.password = password
        return self._repository.update(user)

    def delete_user(self, user_id: int) -> None:
        self._repository.delete(user_id)

from PatronRepository.models.movie import Movie
from PatronRepository.repository.movie_repository import MoviesRepository
from typing import List, Optional

class MovieService:
    """
    Movie Service
    """

    def __init__(self, repository: MoviesRepository):
        self._repository = repository

    def get_all_movies(self) -> List[Movie]:
        return self._repository.get_all()

    def get_movie_by_id(self, movie_id: int) -> Optional[Movie]:
        return self._repository.get_by_id(movie_id)

    def create_movie(self, title: str, genre: str, director: str, release_year: int, rating: float) -> Movie:
        new_movie = Movie(
            title=title,
            genre=genre,
            director=director,
            release_year=release_year,
            rating=rating
        )
        return self._repository.create(new_movie)

    def update_movie(self, movie_id: int, title: str, genre: str, director: str, release_year: int, rating: float) -> bool:
        movie = self._repository.get_by_id(movie_id)
        if movie is None:
            return False
        movie.title = title
        movie.genre = genre
        movie.director = director
        movie.release_year = release_year
        movie.rating = rating
        return self._repository.update(movie)

    def delete_movie(self, movie_id: int) -> None:
        self._repository.delete(movie_id)

from PatronRepository.models.review import Review
from PatronRepository.repository.review_repository import ReviewRepository
from typing import List, Optional

class ReviewService:
    """
    Review Service
    """

    def __init__(self, repository: ReviewRepository):
        self._repository = repository

    def get_all_reviews(self) -> List[Review]:
        return self._repository.get_all()

    def get_review_by_id(self, review_id: int) -> Optional[Review]:
        return self._repository.get_by_id(review_id)

    def create_review(self, movie_id: int, user_id: int, rating: float, review_text: str) -> Review:
        new_review = Review(
            movie_id=movie_id,
            user_id=user_id,
            rating=rating,
            review_text=review_text
        )
        return self._repository.create(new_review)

    def update_review(self, review_id: int, movie_id: int, user_id: int, rating: float, review_text: str) -> bool:
        review = self._repository.get_by_id(review_id)
        if review is None:
            return False
        review.movie_id = movie_id
        review.user_id = user_id
        review.rating = rating
        review.review_text = review_text
        return self._repository.update(review)

    def delete_review(self, review_id: int) -> None:
        self._repository.delete(review_id)

Hasta este punto, ya tenemos todo lo necesario para el funcionamiento básico. Si probamos todas las funciones CRUD, deberían funcionar sin problemas. Vamos a realizar algunas pruebas.

Creación y Listado de Usuarios:

import sqlite3
from PatronRepository.repository.user_repository import UserRepository
from PatronRepository.services.user_service import UserService
from PatronRepository.models.user import User

def main():
    # Conexión a la base de datos
    db_connection = sqlite3.connect("superdb.db")

    try:
        # Repositorio y servicio de usuarios
        user_repository = UserRepository(db_connection)
        user_service = UserService(user_repository)

        # Crear un nuevo usuario
        user = user_service.create_user(
            username='Memo',
            email='[email protected]',
            password='ABC123'
        )
        print(user.username)
        # >>> Memo

        # Obtener todos los usuarios
        users = user_service.get_all_users()

        for user in users:
            print(f"ID: {user.id}, Username: {user.username}, Email: {user.email}, Password: {user.password}, Date Joined: {user.date_joined}")
        # Ejemplo de salida:
        # ID: 1, Username: pepito, Email: [email protected], Password: password123, Date Joined: 2024-07-17
        # ID: 2, Username: juanito, Email: [email protected], Password: password456, Date Joined: 2024-07-17
        # ID: 3, Username: Memo, Email: [email protected], Password: ABC123, Date Joined: None

    finally:
        # Cierre de la conexión a la base de datos
        db_connection.close()

if __name__ == "__main__":
    main()

Es tan sencillo como lo anterior: simplemente con una conexión a la base de datos, creamos un repositorio de usuarios y luego lo inyectamos como dependencia al servicio. El servicio se encarga de todas las tareas. ¡Genial!

Pruebas de Películas y Reseñas:

import sqlite3
from PatronRepository.repository.movie_repository import MoviesRepository
from PatronRepository.repository.review_repository import ReviewRepository
from PatronRepository.services.movie_service import MovieService
from PatronRepository.services.review_service import ReviewService
from PatronRepository.models.movie import Movie
from PatronRepository.models.review import Review

def main():
    # Conexión a la base de datos
    db_connection = sqlite3.connect("superdb.db")

    try:
        # Repositorios y servicios
        movies_repository = MoviesRepository(db_connection)
        reviews_repository = ReviewRepository(db_connection)
        movie_service = MovieService(movies_repository)
        review_service = ReviewService(reviews_repository)

        # Obtener una película por ID
        movie = movie_service.get_movie_by_id(movie_id=5)
        if movie:
            print(f"Pelicula: ID: {movie.id}, Title: {movie.title}, Genre: {movie.genre}, Director: {movie.director}, Release Year: {movie.release_year}, Rating: {movie.rating}")
        else:
            print("Pelicula no encontrada.")

        # Crear una nueva reseña
        review_service.create_review(
            user_id=1,
            movie_id=5,
            review_text="Excelente pelicula",
            rating=9
        )

        # Obtener todas las reseñas
        reviews = review_service.get_all_reviews()
        for review in reviews:
            print(f"Review: ID: {review.id}, Movie ID: {review.movie_id}, User ID: {review.user_id}, Rating: {review.rating}, Text: {review.review_text}, Date: {review.date}")

    finally:
        # Cierre de la conexión a la base de datos
        db_connection.close()

if __name__ == "__main__":
    main()

Añadiendo Métodos Más Concretos

Las operaciones básicas funcionan sin problema, pero ¿qué pasa si necesitas hacer operaciones más concretas con tus datos, como obtener las reviews por usuario o calcular el promedio de reviews de una película? Pues eso lo hacemos en el repositorio concreto y en el servicio de la entidad que lo requiere. Veamos un ejemplo.

Primero, añadimos el método con la query necesaria para obtener la información en el repositorio de reviews.

from sqlite3 import Connection
from typing import List, Optional

from PatronRepository.models.review import Review
from PatronRepository.repository.repository_generic import IRepository

class ReviewRepository(IRepository[Review]):
    def __init__(self, db_connection: Connection):
        self._connection = db_connection

    def get_all(self) -> List[Review]:
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM reviews")
        results = cursor.fetchall()
        return [Review(*result) for result in results]  # Convertir a instancias de Review

    def get_by_id(self, id: int) -> Optional[Review]:
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM reviews WHERE id=?", (id,))
        result = cursor.fetchone()
        return Review(*result) if result is not None else None

    def create(self, item: Review) -> None:
        cursor = self._connection.cursor()
        cursor.execute("""
            INSERT INTO reviews(movie_id, user_id, rating, review_text) VALUES (?, ?, ?, ?)
        """, (item.movie_id, item.user_id, item.rating, item.review_text))
        self._connection.commit()

    def update(self, item: Review) -> None:
        cursor = self._connection.cursor()
        cursor.execute("""
            UPDATE reviews SET movie_id=?, user_id=?, rating=?, review_text=? WHERE id=?
        """, (item.movie_id, item.user_id, item.rating, item.review_text, item.id))
        self._connection.commit()

    def delete(self, id: int) -> None:
        cursor = self._connection.cursor()
        cursor.execute("DELETE FROM reviews WHERE id=?", (id,))
        self._connection.commit()

    def get_by_user_id(self, user_id: int) -> List[Review]:
        """
        Método para obtener todas las reseñas de un usuario.
        """
        cursor = self._connection.cursor()
        cursor.execute("SELECT * FROM reviews WHERE user_id=?", (user_id,))
        results = cursor.fetchall()
        return [Review(*result) for result in results]

    def get_review_score(self, movie_id: int) -> float:
        """
        Método para obtener el promedio de las reseñas de una película.
        """
        cursor = self._connection.cursor()
        cursor.execute("SELECT AVG(rating) FROM reviews WHERE movie_id=?", (movie_id,))
        result = cursor.fetchone()
        return result[0] if result[0] is not None else 0.0

Luego, agregamos la funcionalidad en el servicio de reviews.

from typing import List, Optional
from PatronRepository.models.review import Review
from PatronRepository.repository.review_repository import ReviewRepository

class ReviewService:
    """
    Review Service
    """

    def __init__(self, repository: ReviewRepository):
        self._repository = repository

    def get_all_reviews(self) -> List[Review]:
        """Obtiene todas las reseñas."""
        return self._repository.get_all()

    def get_review_by_id(self, review_id: int) -> Optional[Review]:
        """Obtiene una reseña por ID."""
        return self._repository.get_by_id(review_id)

    def create_review(self, movie_id: int, user_id: int, rating: float, review_text: str) -> Review:
        """Crea una nueva reseña."""
        new_review = Review(movie_id=movie_id, user_id=user_id, rating=rating, review_text=review_text)
        self._repository.create(new_review)
        return new_review

    def update_review(self, review_id: int, movie_id: int, user_id: int, rating: float, review_text: str) -> bool:
        """Actualiza una reseña existente."""
        review = self._repository.get_by_id(review_id)
        if review is None:
            return False
        review.movie_id = movie_id
        review.user_id = user_id
        review.rating = rating
        review.review_text = review_text
        return self._repository.update(review)

    def delete_review(self, review_id: int) -> None:
        """Elimina una reseña."""
        self._repository.delete(review_id)

    def get_reviews_by_user(self, user_id: int) -> List[Review]:
        """
        Obtiene todas las reseñas de un usuario específico.

        :param user_id: ID del usuario.
        :return: Lista de reseñas del usuario.
        """
        return self._repository.get_by_user_id(user_id)

    def get_review_score_by_movie(self, movie_id: int) -> float:
        """
        Obtiene el promedio de las reseñas para una película específica.

        :param movie_id: ID de la película.
        :return: Promedio de las reseñas de la película.
        """
        return self._repository.get_review_score(movie_id)

¡Listo para usar!

# Obtener reseñas por usuario
reviews_by_user = review_service.get_reviews_by_user(user_id=2)

if reviews_by_user:
    for review in reviews_by_user:
        print(review)
else:
    print("No se encontraron reseñas para este usuario.")

# Obtener el puntaje promedio de la película con id 2
review_score = review_service.get_review_score_by_movie(movie_id=2)
print(f"Puntaje promedio de la película con ID 2: {review_score if review_score is not None else 'No disponible'}")

Conclusión

Por fin llegamos al final de esta cosa tan aburrida. El patrón repositorio es muy común y un básico para tener una mejor separación de responsabilidades en nuestros proyectos. Es muy útil para organizar proyectos con diversos tipos de arquitectura. En un futuro, haré una entrada aplicando este patrón a un proyecto con Django o FastAPI.