Servicios para consultar datos

Buen día, comparto mi solución al reto.

Repositorio: https://github.com/cristian-rincon/api-movies

Si les gusta los invito a seguirme en mi github: https://github.com/cristian-rincon

class MovieService:
    def __init__(self, database: Session):
        """Movie service."""
        self.database = database

    def get_movies(self, genre: str = "", year: int = 1900, id: int = 0) -> tuple:
        """Get movies endpoint. If no filters are provided, return all movies.
        
        Args:
            genre (str): Movie genre.
            year (int): Movie year.
            id (int): Movie id.

        Returns:
            tuple: Movies and filtered movies.
        """
        movies = self.database.query(MovieORM).all()
        filtered_movies = []
        if genre:
            filtered_by_genre = get_movies_by_genre(genre, self.database)
            filtered_movies.append(filtered_by_genre)
        elif year:
            filtered_by_year = get_movies_by_year(year, self.database)
            filtered_movies.append(filtered_by_year)
        elif id:
            filtered_by_id = get_movie_by_id(id, self.database)
            filtered_movies.append(filtered_by_id)

        return movies, filtered_movies

solo un apunte, en esta parte se menciona que se coloca self como parametro del metodo get_movies entendí mal que es un metodo de clase, pero en realidad es un metodo de instancia

Adjunto mi reto filtrando por categoria

def get_movies_by_category(self, category):
        result = self.db.query(MovieModel).filter(
            MovieModel.category == category).all()
        return result

Explicacion horrible, no fue un curso sencillo por que el profesor no tiene buena metodologia, mostrar y mostrar configuraciones y codigo no es hacer un curso. 0 y si pudiera colocar un -5 lo hiciera.

Mi solución

En el router

@movie_router.get('/movies/', tags=['movies'], response_model=List[Movie])
def get_movies_by_category(category: str = Query(min_length=5, max_length=20)) -> List[Movie]:
    db = Session()
    result = MovieService(db).get_movie_by_category(category)
    if not result:
        return JSONResponse(status_code=404, content={'message': movie_not_found})
    return JSONResponse(status_code=200, content=jsonable_encoder(result))

Mi solución

El servicio

    def get_movie_by_category(self, category):
        result = self.db.query(MovieModel).filter(MovieModel.category == category).all()
        return result

Mi solucion al reto

async def get_movies_by_category(self, category: str):
      with Session(engine) as session:
          statement = select(Movie).where(Movie.category == category)
          movies = session.exec(statement).all()
      return movies

Clase asociada al servicio de movie

class MovieService():
    def __init__(self, db) -> None:
        self.db = db
    
    def get_movies(self):
        result = self.db.query(MovieModel).all()
        return result
    
    def get_movie(self, id):
        return self.db.query(MovieModel).filter(MovieModel.id == id).first()
    
    def get_movie_by_category(self, category):
        return self.db.query(MovieModel).filter(MovieModel.category == category).all()

Mi solución al reto

movie/services

def get_movie_by_category(self,category):
        result=self.db.query(MovieModel).filter(MovieModel.category==category).all()
        return result

movie/routers

def get_movies_by_category(category: str = Query(min_length=5, max_length=15)) -> List[Movie]:
    db=Session()
    result= MovieService(db).get_movie_by_category(category)
    if not result:
         return JSONResponse(status_code=404, content={"Message":"Not found"})
    return JSONResponse(status_code=200,content=jsonable_encoder(result))