File size: 3,620 Bytes
67aa1ca | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | """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.")
|