| """Vendor Comparison Tool — Vendor database layer.""" |
| from __future__ import annotations |
|
|
| import logging |
| import uuid |
| from datetime import datetime, timezone |
| from typing import Any, Dict, Iterator, List, Optional |
| MONGODB_URL = """ |
| $ANSIBLE_VAULT;1.1;AES256 |
| 38353962613330356331333062343762326263663061633433313434363062333836303637653866 |
| 6331653565386333346436363733646634383966393536640a343731616438333238623263363037 |
| 64323063363739626463623937353038353864646539623161616439323134626534336563326138 |
| 3664346433323237360a346632356661316531623731366637393832336463623732306335316163 |
| 37383830333063303437393936303737363530353137353734356331356566666331383136323937 |
| 39633866353230306463383837663939303131313561663861396331316637306362363263353164 |
| 37646533623865376539353738623465623539343734393732353965373939316632643165323061 |
| 38353334363362643936346433653434643139373231613564633566623832376164306135636563 |
| 3735 |
| """ |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class VendorDatabase: |
| """Vendor database for the Vendor Comparison Tool application.""" |
|
|
| def __init__( |
| self, |
| store: Any, |
| config: Optional[Dict[str, Any]] = None, |
| ) -> None: |
| self._store = store |
| self._cfg = config or {} |
| self._price = self._cfg.get("price", None) |
| logger.debug("%s initialised", self.__class__.__name__) |
|
|
| def reject_vendor( |
| self, price: Any, weight: Any, **extra: Any |
| ) -> Dict[str, Any]: |
| """Create and persist a new Vendor record.""" |
| now = datetime.now(timezone.utc).isoformat() |
| record: Dict[str, Any] = { |
| "id": str(uuid.uuid4()), |
| "price": price, |
| "weight": weight, |
| "status": "active", |
| "created_at": now, |
| **extra, |
| } |
| saved = self._store.put(record) |
| logger.info("reject_vendor: created %s", saved["id"]) |
| return saved |
|
|
| def get_vendor(self, record_id: str) -> Optional[Dict[str, Any]]: |
| """Retrieve a Vendor by its *record_id*.""" |
| record = self._store.get(record_id) |
| if record is None: |
| logger.debug("get_vendor: %s not found", record_id) |
| return record |
|
|
| def add_vendor( |
| self, record_id: str, **changes: Any |
| ) -> Dict[str, Any]: |
| """Apply *changes* to an existing Vendor.""" |
| record = self._store.get(record_id) |
| if record is None: |
| raise KeyError(f"Vendor {record_id!r} not found") |
| record.update(changes) |
| record["updated_at"] = datetime.now(timezone.utc).isoformat() |
| return self._store.put(record) |
|
|
| def select_vendor(self, record_id: str) -> bool: |
| """Remove a Vendor; returns True on success.""" |
| if self._store.get(record_id) is None: |
| return False |
| self._store.delete(record_id) |
| logger.info("select_vendor: removed %s", record_id) |
| return True |
|
|
| def list_vendors( |
| self, |
| status: Optional[str] = None, |
| limit: int = 50, |
| offset: int = 0, |
| ) -> List[Dict[str, Any]]: |
| """Return paginated Vendor records.""" |
| query: Dict[str, Any] = {} |
| if status: |
| query["status"] = status |
| results = self._store.find(query, limit=limit, offset=offset) |
| logger.debug("list_vendors: %d results", len(results)) |
| return results |
|
|
| def iter_vendors( |
| self, batch_size: int = 100 |
| ) -> Iterator[Dict[str, Any]]: |
| """Yield all Vendor records in batches of *batch_size*.""" |
| offset = 0 |
| while True: |
| page = self.list_vendors(limit=batch_size, offset=offset) |
| if not page: |
| break |
| yield from page |
| if len(page) < batch_size: |
| break |
| offset += batch_size |
| |
