Spaces:
Sleeping
Sleeping
| """Servicio base con operaciones CRUD genéricas sobre una colección. | |
| Las entidades concretas heredan de :class:`BaseService` y añaden su | |
| lógica específica (validación de referencias, marcas de tiempo, etc.). | |
| Trabajar contra esta base mantiene el sistema genérico y fácil de | |
| extender: añadir una nueva entidad se reduce a crear un servicio que | |
| herede de aquí. | |
| """ | |
| from __future__ import annotations | |
| from typing import Any | |
| from bson import ObjectId | |
| from pymongo import ReturnDocument | |
| from pymongo.asynchronous.collection import AsyncCollection | |
| from pymongo.asynchronous.database import AsyncDatabase | |
| from app.database.objectid import to_object_id | |
| from app.exceptions import NotFoundError | |
| Document = dict[str, Any] | |
| class BaseService: | |
| """CRUD genérico sobre una colección de MongoDB.""" | |
| #: Nombre de la colección gestionada por el servicio. | |
| collection_name: str | |
| #: Nombre del recurso, usado en los mensajes de error. | |
| recurso: str = "Recurso" | |
| def __init__(self, db: AsyncDatabase) -> None: | |
| self.db = db | |
| self.collection: AsyncCollection = db[self.collection_name] | |
| # -- Lectura --------------------------------------------------------- | |
| async def get(self, id_: str | ObjectId) -> Document: | |
| """Obtiene un documento por su id o lanza :class:`NotFoundError`.""" | |
| document = await self.collection.find_one({"_id": to_object_id(id_)}) | |
| if document is None: | |
| raise NotFoundError(f"{self.recurso} no encontrado.") | |
| return document | |
| async def find( | |
| self, | |
| filters: Document | None = None, | |
| *, | |
| skip: int = 0, | |
| limit: int = 50, | |
| sort: list[tuple[str, int]] | None = None, | |
| ) -> list[Document]: | |
| """Devuelve una lista de documentos según el filtro indicado.""" | |
| cursor = self.collection.find(filters or {}) | |
| if sort: | |
| cursor = cursor.sort(sort) | |
| cursor = cursor.skip(skip).limit(limit) | |
| return [document async for document in cursor] | |
| async def count(self, filters: Document | None = None) -> int: | |
| """Cuenta los documentos que cumplen el filtro.""" | |
| return await self.collection.count_documents(filters or {}) | |
| async def exists(self, id_: str | ObjectId) -> bool: | |
| """Indica si existe un documento con el id dado.""" | |
| return ( | |
| await self.collection.count_documents({"_id": to_object_id(id_)}, limit=1) | |
| > 0 | |
| ) | |
| # -- Escritura ------------------------------------------------------- | |
| async def _insert(self, document: Document) -> Document: | |
| """Inserta un documento y lo devuelve ya persistido.""" | |
| result = await self.collection.insert_one(document) | |
| return await self.get(result.inserted_id) | |
| async def _apply_update(self, id_: str | ObjectId, changes: Document) -> Document: | |
| """Aplica ``$set`` con los cambios y devuelve el documento.""" | |
| oid = to_object_id(id_) | |
| if not changes: | |
| return await self.get(oid) | |
| document = await self.collection.find_one_and_update( | |
| {"_id": oid}, | |
| {"$set": changes}, | |
| return_document=ReturnDocument.AFTER, | |
| ) | |
| if document is None: | |
| raise NotFoundError(f"{self.recurso} no encontrado.") | |
| return document | |
| async def delete(self, id_: str | ObjectId) -> None: | |
| """Elimina un documento por su id.""" | |
| result = await self.collection.delete_one({"_id": to_object_id(id_)}) | |
| if result.deleted_count == 0: | |
| raise NotFoundError(f"{self.recurso} no encontrado.") | |