commit 9283ded5ae56e613bb5304720af37cfc7b5981bb Author: Andrey Epifancev Date: Thu Sep 10 18:05:48 2026 +0400 Initial commit: DKB parsers service FastAPI service that consumes NATS `files.uploaded` events and parses uploaded documents from MinIO/S3 into legal entity, company group and RF office schemas. Co-Authored-By: Claude Opus 5 (1M context) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..02f7e32 --- /dev/null +++ b/.gitignore @@ -0,0 +1,35 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +dist/ +*.egg-info/ +.eggs/ + +# Virtual environments +.venv/ +venv/ +env/ +ENV/ + +# Environment / secrets +.env +.env.* +!.env.example + +# Tooling caches +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +htmlcov/ + +# IDE +.idea/ +.vscode/ + +# OS +.DS_Store diff --git a/main.py b/main.py new file mode 100644 index 0000000..290910e --- /dev/null +++ b/main.py @@ -0,0 +1,58 @@ +from contextlib import asynccontextmanager + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from miniopy_async import Minio +from src.core.acync_cache import AsyncInMemoryCache +from src.core.poll_worker import PollWorker +from src.core.config import settings +from src.core.clients import S3Storage +from nats.aio.client import Client as NATS +import asyncio +from miniopy_async.error import S3Error + +import json + +from src.core.nats_callback import build_file_handler +from src.core.nats_callback_pdf import build_file_handler_pdf + + +@asynccontextmanager +async def lifespan(app): + s3_client = Minio(settings.MINIO_URL, settings.MINIO_ADMIN_LOGIN, settings.MINIO_ADMIN_PASSWORD, secure=False) + storage = S3Storage(client=s3_client, bucket_name=settings.MINIO_BUCKET) + app.state.storage = storage + nc = NATS() + await nc.connect("nats://localhost:4222", max_reconnect_attempts=5) + # cache = AsyncInMemoryCache() + # poller = PollWorker(nc, storage, cache) + # await poller.start() + + handler_cb = build_file_handler(nc, storage, bucket_name=settings.MINIO_BUCKET) + # handler_cb = build_file_handler_pdf(nc, storage, settings.MINIO_BUCKET, cache) + + sub = await nc.subscribe('files.uploaded', queue='xlsxparsing', cb=handler_cb) + app.state.nats = nc + app.state.nats_subs = [sub] + + try: + yield + finally: + # await poller.stop() + for sub in app.state.nats_subs: + await sub.unsubscribe() + + await nc.close() + +app = FastAPI(lifespan=lifespan) + +@app.get('/healthcheck') +async def health(): + return { + 'status': 'alive' + } + +app.add_middleware( + CORSMiddleware, + allow_credentials=['*'], allow_headers=['*'], allow_methods=['*'], allow_origins=['*'] +) \ No newline at end of file diff --git a/src/core/acync_cache.py b/src/core/acync_cache.py new file mode 100644 index 0000000..b50cfbe --- /dev/null +++ b/src/core/acync_cache.py @@ -0,0 +1,43 @@ +import asyncio +from datetime import datetime, timedelta +import uuid + +from .types import FileDictItem + +class AsyncInMemoryCache: + def __init__(self): + self.mapping: dict[str, FileDictItem] = {} + self._lock = asyncio.Lock() + + async def add(self, job_id: uuid.UUID, file_id: uuid.UUID) -> FileDictItem: + now = datetime.now() + expiration_date = datetime.now() + timedelta(minutes=5) + + new_item = { + 'started_at': now, + 'expires_at': expiration_date, + 'file_id': str(file_id), + 'job_id': str(job_id) + } + async with self._lock: + self.mapping[str(job_id)] = new_item + print("ADDED IN CACHE FROM OUTSIDE") + return new_item + + async def get(self, job_id: uuid.UUID): + async with self._lock: + return self.mapping.get(str(job_id)) + + async def remove(self, source_link: str) -> None: + async with self._lock: + self._items.pop(source_link, None) + + + async def get_active_items(self) -> list[FileDictItem]: + print("ACTIVE ITEMS") + print(self.mapping) + async with self._lock: + return [ + item + for item in self.mapping.values() + ] \ No newline at end of file diff --git a/src/core/clients.py b/src/core/clients.py new file mode 100644 index 0000000..1a0d754 --- /dev/null +++ b/src/core/clients.py @@ -0,0 +1,14 @@ +from miniopy_async import Minio +from io import BytesIO + +class S3Storage: + def __init__(self, client: Minio, bucket_name: str): + self.client = client + self.bucket = bucket_name + + async def put_object(self, obj_name, file_data): + try: + stream = BytesIO(file_data) + await self.client.put_object(self.bucket, obj_name, stream, length=len(file_data)) + except: + raise \ No newline at end of file diff --git a/src/core/config.py b/src/core/config.py new file mode 100644 index 0000000..653525e --- /dev/null +++ b/src/core/config.py @@ -0,0 +1,12 @@ +from pydantic_settings import BaseSettings, SettingsConfigDict + +class Settings(BaseSettings): + DATABASE_URL: str + MINIO_URL: str + MINIO_ADMIN_LOGIN: str + MINIO_ADMIN_PASSWORD: str + MINIO_BUCKET: str + NATS_URL: str + model_config = SettingsConfigDict(env_file=".env", extra="ignore") + +settings = Settings() \ No newline at end of file diff --git a/src/core/nats_callback.py b/src/core/nats_callback.py new file mode 100644 index 0000000..9fb9224 --- /dev/null +++ b/src/core/nats_callback.py @@ -0,0 +1,100 @@ +import asyncio +import json +import io +import pandas as pd +from datetime import datetime, timedelta + +from nats.aio.client import Client as NATS +from miniopy_async.error import S3Error + +from .clients import S3Storage +from .acync_cache import AsyncInMemoryCache +from .types import FileDictItem + + +def create_json_file(job_id, leg_entity_id, stored_file_id): + file_data = { + 'recognition_job_id': job_id, + 'legal_entity_id': leg_entity_id, + 'stored_file_id': stored_file_id, + 'period_quarter': 'Q3', + 'period_year': 2024, + 'rows': [ + { + 'account_code': '70.01', + 'account_root': '70', + 'counterparty_name': 'ООО Компания', + # 'counterparty_id': 'c161c259-3a80-45aa-af7b-33cd5a3d2731', + 'counterparty_inn': '34146448377', + 'opening_debit': 45.87, + 'opening_credit': 65.98, + 'turnover_debit': 32.70, + 'turnover_credit': 643.98, + 'closing_debit': 43.87, + 'closing_credit': 45.59 + }, + { + 'account_code': '70.02', + 'account_root': '70', + 'counterparty_name': 'ООО Компания', + # 'counterparty_id': 'c161c259-3a80-45aa-af7b-33cd5a3d2731', + 'counterparty_inn': '34146448377', + 'opening_debit': 5.87, + 'opening_credit': 62.38, + 'turnover_debit': 31.40, + 'turnover_credit': 63.98, + 'closing_debit': 4.83, + 'closing_credit': 4.59 + } + ] + } + return json.dumps(file_data).encode('utf-8') + +def build_file_handler(nc: NATS, storage: S3Storage, bucket_name: str): + async def handler(msg): + print("GOT MESSAGE") + print(msg) + await asyncio.sleep(3) + response_payload = { + 'status': None, + 'data': None + } + try: + payload = json.loads(msg.data.decode('utf-8')) + # print("PAYLOAD", payload) + s3_key = payload['s3_key'] + if not(s3_key.lower().endswith('xlsx') or s3_key.lower().endswith('xls')): + return + file_id = payload['file_id'] + legal_entity_id = payload['legal_entity_id'] + + response = await storage.client.get_object(bucket_name, s3_key) # заменить на storage.get_object('gdffd') + file = await response.content.read() + bio = io.BytesIO(file) + df = pd.read_excel(bio) + result_bytes = create_json_file(payload['job_id'], legal_entity_id, file_id) + await storage.put_object(f'{file_id}.json', file_data=result_bytes) + + response_payload['status'] = 'success' + response_payload['data'] = { + # 'result_s3_path': f'{file_id}.json', + 'result_s3_path': None, + **payload + } + + + + # storage.put_object() + # file + # print("RESPONSE", response) + except S3Error: + print("NO S3 FILE") + response_payload['status'] = 'error' + response_payload['data'] = { + **payload + } + + + await nc.publish('files.parsed', json.dumps(response_payload).encode('utf-8')) + + return handler diff --git a/src/core/nats_callback_pdf.py b/src/core/nats_callback_pdf.py new file mode 100644 index 0000000..ff7e8c2 --- /dev/null +++ b/src/core/nats_callback_pdf.py @@ -0,0 +1,143 @@ +import asyncio +import json +import io +import pandas as pd +from datetime import datetime, timedelta + +from nats.aio.client import Client as NATS +from miniopy_async.error import S3Error + +from .clients import S3Storage +from .acync_cache import AsyncInMemoryCache +from .types import FileDictItem + + + +def create_json_file(job_id, file_id): + file_data = { + "job_id": job_id, + "files": [ + { + 'file_uid': file_id, + "error": { + "code": "", + "message": "" + }, + "data_ocr": { + "label": { + 'form_name': 's' + }, + "organization": {}, + "inn": {}, + "units": {}, + "date": {'form_name': '01.06.2026', 'confidence': 100.0}, + "table_structure": [ + { + "label": { + "form_name": "Бухгалтерский баланс", + "confidence": 100.0 + }, + "rows": [ + { + "row_index": 0, + "cells": [ + { + "column_index": 0, + "value": "Наименование показателя 2", + "confidence": 100.0 + }, + { + "column_index": 1, + "value": "Код", + "confidence": 100.0 + }, + { + "column_index": 2, + "value": "За первое полугодие", + "confidence": 100.0 + } + ] + }, + { + "row_index": 1, + "cells": [ + { + "column_index": 0, + "value": "Наименование показателя 2", + "confidence": 100.0 + }, + { + "column_index": 1, + "value": "1100", + "confidence": 100.0 + }, + { + "column_index": 2, + "value": 984, + "confidence": 100.0 + } + ] + } + ] + } + ] + } + } + ] + } + + +def build_file_handler_pdf(nc: NATS, storage: S3Storage, bucket_name: str, cache: AsyncInMemoryCache): + async def handler(msg): + + + print("GOT MESSAGE") + print(msg) + await asyncio.sleep(3) + response_payload = { + 'status': None, + 'data': None + } + try: + payload = json.loads(msg.data.decode('utf-8')) + print("PAYLOAD", payload) + s3_key = payload['s3_key'] + if (s3_key.lower().endswith('xlsx') or s3_key.lower().endswith('xls')): + return + file_id = payload['file_id'] + job_id = payload['job_id'] + legal_entity_id = payload['legal_entity_id'] + + response = await storage.client.get_object(bucket_name, s3_key) # заменить на storage.get_object('gdffd') + file = await response.content.read() + with open(f'/home/sokratmillman/projects/work-test-processor/files/input/{s3_key}', 'wb') as f: + f.write(file) + await cache.add(str(job_id), str(file_id)) + print("ADDED to CACHE", str(job_id), str(file_id)) + # bio = io.BytesIO(file) + # df = pd.read_excel(bio) + # result_bytes = create_json_file(job_id) + # await storage.put_object(f'{file_id}.json', file_data=result_bytes) + + # response_payload['status'] = 'success' + # response_payload['data'] = { + # 'result_s3_path': f'{file_id}.json', + # **payload + # } + + + + # storage.put_object() + # file + # print("RESPONSE", response) + except S3Error: + print("NO S3 FILE") + response_payload['status'] = 'error' + response_payload['data'] = { + **payload + } + + + await nc.publish('files.parsed', json.dumps(response_payload).encode('utf-8')) + + return handler \ No newline at end of file diff --git a/src/core/poll_worker.py b/src/core/poll_worker.py new file mode 100644 index 0000000..bce0533 --- /dev/null +++ b/src/core/poll_worker.py @@ -0,0 +1,107 @@ +import asyncio +import json +import os +from datetime import datetime, timedelta + +from nats.aio.client import Client as NATS + +from .clients import S3Storage +from .acync_cache import AsyncInMemoryCache +from .types import FileDictItem +class PollWorker: + def __init__(self, nats_client: NATS, storage: S3Storage, tracker: AsyncInMemoryCache): + self.task: asyncio.Task | None = None + self.stop_event = asyncio.Event() + self.nats_client = nats_client + self.tracker = tracker + self.interval = 5 + self.storage = storage + + + async def start(self): + self.stop_event.clear() + self.task = asyncio.create_task(self.run()) + + async def stop(self): + self.stop_event.set() + if self.task: + self.task.cancel() + try: + await self.task + except asyncio.CancelledError: + pass + + async def run(self): + # logger.info("Pending publisher started") + print("PUBLISHER STARTED") + while not self.stop_event.is_set(): + try: + await self.poll_once() + except Exception: + # logger.exception("Poll iteration failed") + pass + + try: + await asyncio.wait_for( + self.stop_event.wait(), + timeout=self.interval, + ) + except asyncio.TimeoutError: + pass + + # logger.info("Pending publisher stopped") + print("PUBLISHER STOPPED") + + async def poll_once(self): + print("POLLING FILES") + try: + jobs: list[FileDictItem] = await self.tracker.get_active_items() + print("GOT ") + except Exception as e: + print("FAILED TO CLAIN ROWS") + print(e) + + all_current_files = os.listdir('/home/sokratmillman/projects/work-test-processor/files/output/') + print("ALL FILES", all_current_files) + + for job in jobs: + print("ITERATE OVER FILES") + search_fname = f"{job['file_id']}.json" + print("SEARCH NAME", search_fname) + now = datetime.now() + try: + if (job['expires_at'] + timedelta(seconds=2*self.interval) > now): + try: + await self.publish_nats_expired(job['file_id']) + await self.tracker.remove(job['job_id']) + print("PUBLISHED TIMEOUT ERROR") + except Exception as e: + print("ERROR IN TIMEOUT HANDLIING") + except Exception as e: + print("ERROR", e) + + + print("BEFORE TRY") + print("JOB", job) + try: + print("PUBLISHING") + # await self.publish_to_nats(job["row_payload"]) + print("PUBLISHED SUCESSFULLY") + except Exception as e: + # logger.exception("Failed to publish row id=%s", row_id) + print("FAILED TO PUBLISH", job['job_id']) + print(e) + + # await self.mark_queued(job.job_id) + + + async def publish_to_nats(self, payload): + await self.nats_client.publish('files.parsed', json.dumps(payload).encode("utf-8")) + + + async def publish_nats_expired(self, job_id): + error_payload = { + 'status': 'error', + 'error_message': 'timeout error' + } + await self.nats_client.publish('files.parsed', json.dumps(error_payload).encode("utf-8")) \ No newline at end of file diff --git a/src/core/types.py b/src/core/types.py new file mode 100644 index 0000000..1cbde60 --- /dev/null +++ b/src/core/types.py @@ -0,0 +1,12 @@ +from dataclasses import dataclass +from datetime import datetime + +@dataclass +class FileDictItem: + started_at: datetime + expires_at: datetime + file_id: str + job_id: str + +# @dataclass +# class OsvRow: \ No newline at end of file diff --git a/src/schemas/__init__.py b/src/schemas/__init__.py new file mode 100644 index 0000000..18579b6 --- /dev/null +++ b/src/schemas/__init__.py @@ -0,0 +1,3 @@ +from .legal_entity import LegalEntityPostRequest, LegalEntityPostResponse +from .company_group import CompanyGroupPostRequest, CompanyGroupPostResponse +from .rf_office import RfOfficePostRequest, RfOfficePostResponse \ No newline at end of file diff --git a/src/schemas/company_group.py b/src/schemas/company_group.py new file mode 100644 index 0000000..76b7f94 --- /dev/null +++ b/src/schemas/company_group.py @@ -0,0 +1,19 @@ +from pydantic import BaseModel, ConfigDict +import uuid +from datetime import datetime, date + +class CompanyGroupPostRequest(BaseModel): + crm_id: str + name: str + base_regional_office_id: uuid.UUID + updated_by: str + +class CompanyGroupPostResponse(BaseModel): + id: uuid.UUID + crm_id: str + name: str + base_regional_office_id: uuid.UUID + valid_from: datetime + updated_by: str + created_at: datetime + model_config = ConfigDict(from_attributes=True) \ No newline at end of file diff --git a/src/schemas/enums.py b/src/schemas/enums.py new file mode 100644 index 0000000..3725ba3 --- /dev/null +++ b/src/schemas/enums.py @@ -0,0 +1,65 @@ +from enum import StrEnum + +class LegalEntityType(StrEnum): + IP = 'ИП' + UL = 'ЮЛ' + +class RelatedGroupKind(StrEnum): + GSK = 'gsk' + GSZ = 'gsz' + +class AccountKind(StrEnum): + A = 'A' + P = 'P' + AP = 'AP' + + +class DocumentType(StrEnum): + BUH_OT = "Бухгалтерская отчётность" + OSV = "ОСВ" + OFR = "ОФР" + OTHER = "Прочий документ" + +class FileType(StrEnum): + XLSX = "XLSX" + PDF = "PDF" + PNG = "PNG" + JPG = "JPG" + +class RecognitionPipeline(StrEnum): + TABULAR = "tabular" + IMAGE = "image" + +class PeriodQuarter(StrEnum): + Q1 = "Q1" + Q2 = "Q2" + Q3 = "Q3" + Q4 = "Q4" + +class FileStatus(StrEnum): + UPLOADED = "Загружен" + PROCESSING = "Обработка" + CHECK="Требуется проверка" + ERROR="Ошибка" + +class RecognitionStatus(StrEnum): + QUEUED = "queued" + PROCESSING = "processing" + DONE = "done" + ERROR = "error" + +class CounterpartyType(StrEnum): + GK = "ГК" + GSK = "ГСК" + EXTERNAL = "Внешний" + +class StatementType(StrEnum): + BALANCE = "balance" + INCOME = "income" + +class FslRowKind(StrEnum): + SECTION = "section" + ROW = "row" + SUB = "sub" + TOTAL = "total" + BALANCE = "balance" \ No newline at end of file diff --git a/src/schemas/legal_entity.py b/src/schemas/legal_entity.py new file mode 100644 index 0000000..667f818 --- /dev/null +++ b/src/schemas/legal_entity.py @@ -0,0 +1,25 @@ +from pydantic import BaseModel, ConfigDict, Field +import uuid +from datetime import datetime, date +from .enums import LegalEntityType + +class LegalEntityPostRequest(BaseModel): + inn: str = Field(min_length=10, max_length=12, pattern=r'^\d+$') + entity_type: LegalEntityType | None + name: str + company_group_id: uuid.UUID + regional_office_id: uuid.UUID + close_date: date + updated_by: str + +class LegalEntityPostResponse(BaseModel): + id: uuid.UUID + inn: str + entity_type: LegalEntityType | None + name: str + company_group_id: uuid.UUID + regional_office_id: uuid.UUID + close_date: date + updated_by: str + created_at: datetime + model_config = ConfigDict(from_attributes=True) diff --git a/src/schemas/rf_office.py b/src/schemas/rf_office.py new file mode 100644 index 0000000..65cd6f0 --- /dev/null +++ b/src/schemas/rf_office.py @@ -0,0 +1,12 @@ +from pydantic import BaseModel, ConfigDict +import uuid + +class RfOfficePostRequest(BaseModel): + name: str + is_active: bool = True + +class RfOfficePostResponse(BaseModel): + id: uuid.UUID + name: str + is_active: bool + model_config = ConfigDict(from_attributes=True) \ No newline at end of file diff --git a/src/services/__init__.py b/src/services/__init__.py new file mode 100644 index 0000000..f2f4ebe --- /dev/null +++ b/src/services/__init__.py @@ -0,0 +1,4 @@ +from .company_group_service import CompanyGroupService +from .file_service import FileService +from .legal_entity_service import LegalEntityService +from .rf_office_service import RfOfficeService \ No newline at end of file diff --git a/src/services/company_group_service.py b/src/services/company_group_service.py new file mode 100644 index 0000000..cbffe8b --- /dev/null +++ b/src/services/company_group_service.py @@ -0,0 +1,30 @@ +import uuid + +from fastapi import UploadFile +from sqlalchemy.ext.asyncio import AsyncSession +from src.repositories.company_group_repo import CompanyGroupRepository +from src.common.exceptions import NotFoundError, CouldntCreateError + +class CompanyGroupService: + def __init__(self, session: AsyncSession, com_group_repo: CompanyGroupRepository): + self.com_group_repo = com_group_repo + self.session = session + + async def create_company_group(self, data): + try: + company_group = await self.com_group_repo.insert_one(data.model_dump()) + await self.session.commit() + return company_group + except Exception as e: + await self.session.rollback() + raise + + async def get_all_groups(self): + try: + company_groups = await self.com_group_repo.get_all() + if len(company_groups) == 0: + raise NotFoundError('No company groups found') + return company_groups + except Exception as e: + raise + \ No newline at end of file diff --git a/src/services/file_service.py b/src/services/file_service.py new file mode 100644 index 0000000..f69b919 --- /dev/null +++ b/src/services/file_service.py @@ -0,0 +1,72 @@ +import uuid + +from fastapi import UploadFile +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.inspection import inspect + +from src.core.clients import S3Storage +from src.models.enums import FileStatus, FileType + +def get_dict(obj): + return {c.key: getattr(obj, c.key) for c in inspect(obj).mapper.column_attrs} + +class FileService: + def __init__(self, storage: S3Storage, session: AsyncSession, file_repository, recognition_job_repo, nats_conn): + self.storage = storage + self.file_repo = file_repository + self.rec_job_repo = recognition_job_repo + self.nats_conn = nats_conn + self.session = session + + async def save_file(self, file: UploadFile, legal_entity_id: uuid.UUID, quarter: str, year: int): + try: + db_payload = { + 'original_name': file.filename, + 'legal_entity_id': legal_entity_id, + 'period_quarter': quarter, + 'period_year': year, + 'size_bytes': file.size, + } + file_ext = file.filename.lower().split('.')[-1] + if (file.filename.endswith('xlsx')): + db_payload['file_type'] = FileType.XLSX + elif (file.filename.endswith('pdf')): + db_payload['file_type'] = FileType.PDF + elif (file.filename.endswith('png')): + db_payload['file_type'] = FileType.PNG + elif (file.filename.endswith('jpg')): + db_payload['file_type'] = FileType.JPG + new_file = await self.file_repo.insert_one(db_payload) + + s3_key = f'{str(new_file.id)}.{file_ext}' + await self.storage.put_object(s3_key, await file.read()) + new_file.s3_key = s3_key + new_file.status = FileStatus.UPLOADED + + rec_job_db_payload = { + 'file_id': new_file.id, + 'attempt': 1 + } + rec_job = await self.rec_job_repo.insert_one(rec_job_db_payload) + new_file.current_recognition_job_id = rec_job.id + + await self.session.commit() + + print(get_dict(rec_job)) + print(get_dict(new_file)) + return {'file': get_dict(new_file), 'rec_job': get_dict(rec_job)} + + + except Exception as e: + await self.session.rollback() + print("ERROR") + print(e) + + raise + + async def _save_to_storage(self, file: UploadFile, file_id: uuid.UUID): + print("saving one file") + print(file) + + def _generate_file_ids(): + pass \ No newline at end of file diff --git a/src/services/legal_entity_service.py b/src/services/legal_entity_service.py new file mode 100644 index 0000000..fe96934 --- /dev/null +++ b/src/services/legal_entity_service.py @@ -0,0 +1,28 @@ +import uuid + +from fastapi import UploadFile +from sqlalchemy.ext.asyncio import AsyncSession +from src.common.exceptions import NotFoundError +from src.repositories.legal_entity_repo import LegalEntityRepository, LegalEntity + +class LegalEntityService: + def __init__(self, session: AsyncSession, le_repo: LegalEntityRepository): + self.le_repo = le_repo + self.session = session + + async def create_legal_entity(self, data): + try: + legal_entity: LegalEntity = await self.le_repo.insert_one(data.model_dump()) + await self.session.commit() + return legal_entity + except Exception as e: + await self.session.rollback() + raise + async def get_all_offices(self): + try: + ref_offices = await self.le_repo.get_all() + if len(ref_offices) == 0: + raise NotFoundError('No legal entities found') + return ref_offices + except Exception as e: + raise \ No newline at end of file diff --git a/src/services/rf_office_service.py b/src/services/rf_office_service.py new file mode 100644 index 0000000..eb52f44 --- /dev/null +++ b/src/services/rf_office_service.py @@ -0,0 +1,30 @@ +import uuid + +from fastapi import UploadFile +from sqlalchemy.ext.asyncio import AsyncSession +from src.repositories import RfRepository +from src.common.exceptions import NotFoundError, CouldntCreateError + +class RfOfficeService: + def __init__(self, session: AsyncSession, ref_office_repo: RfRepository): + self.ref_office_repo = ref_office_repo + self.session = session + + async def create_ref_office(self, data): + try: + ref_office = await self.ref_office_repo.insert_one(data.model_dump()) + await self.session.commit() + return ref_office + except Exception as e: + await self.session.rollback() + raise + + async def get_all_offices(self): + try: + ref_offices = await self.ref_office_repo.get_all() + if len(ref_offices) == 0: + raise NotFoundError('No company groups found') + return ref_offices + except Exception as e: + raise + \ No newline at end of file