diff --git a/api/alembic/env.py b/api/alembic/env.py index 89ae547..65e43ee 100644 --- a/api/alembic/env.py +++ b/api/alembic/env.py @@ -87,7 +87,7 @@ def run_migrations_offline() -> None: target_metadata=target_metadata, literal_binds=True, dialect_opts={"paramstyle": "named"}, - version_table_schema="dkb", + version_table_schema=db_schema, include_schemas=True, ) diff --git a/api/alembic/versions/c94eb1503a26_add_osv_counterparty_alias_flag.py b/api/alembic/versions/c94eb1503a26_add_osv_counterparty_alias_flag.py index 344be9e..7cf1813 100644 --- a/api/alembic/versions/c94eb1503a26_add_osv_counterparty_alias_flag.py +++ b/api/alembic/versions/c94eb1503a26_add_osv_counterparty_alias_flag.py @@ -4,6 +4,7 @@ Revision ID: c94eb1503a26 Revises: d614b414a32f """ from alembic import op +from src.core.config import settings revision = "c94eb1503a26" down_revision = "d614b414a32f" @@ -11,12 +12,19 @@ branch_labels = None depends_on = None +def _quoted_db_schema() -> str: + return op.get_bind().dialect.identifier_preparer.quote_schema( + settings.OPENBAO.DATABASE.DATABASE_SCHEMA + ) + + def upgrade() -> None: + schema = _quoted_db_schema() # PostgreSQL requires recreation when RETURNS TABLE changes. # No CASCADE: dependent database objects must not be silently removed. - op.execute("DROP FUNCTION dkb.get_osv_entries_for_group(uuid, integer, text)") - op.execute(""" - CREATE OR REPLACE FUNCTION dkb.get_osv_entries_for_group( + op.execute(f"DROP FUNCTION {schema}.get_osv_entries_for_group(uuid, integer, text)") + op.execute(f""" + CREATE OR REPLACE FUNCTION {schema}.get_osv_entries_for_group( p_company_group_id uuid, p_period_year integer, p_period_quarter text @@ -94,8 +102,8 @@ def upgrade() -> None: rj.finished_at, EXISTS ( SELECT 1 - FROM dkb.related_group_member AS member - JOIN dkb.related_group AS rg + FROM {schema}.related_group_member AS member + JOIN {schema}.related_group AS rg ON rg.id = member.related_group_id WHERE member.legal_entity_id = oe.legal_entity_id AND member.is_current IS TRUE @@ -103,26 +111,26 @@ def upgrade() -> None: AND rg.kind = 'GSK' ) AS is_gsk, le.company_group_id = p_company_group_id AS is_gk - FROM dkb.osv_entry AS oe - LEFT JOIN dkb.legal_entity AS counterparty_by_inn + FROM {schema}.osv_entry AS oe + LEFT JOIN {schema}.legal_entity AS counterparty_by_inn ON oe.counterparty_id IS NULL AND counterparty_by_inn.inn = NULLIF(BTRIM(oe.counterparty_inn), '') - LEFT JOIN dkb.legal_entity_alias AS counterparty_by_alias + LEFT JOIN {schema}.legal_entity_alias AS counterparty_by_alias ON oe.counterparty_id IS NULL AND counterparty_by_inn.id IS NULL AND counterparty_by_alias.normalized_alias = NULLIF(BTRIM(oe.counterparty_name), '') - LEFT JOIN dkb.chart_of_account AS coa ON coa.id = oe.account_id - JOIN dkb.stored_file AS sf ON sf.id = oe.stored_file_id - JOIN dkb.legal_entity AS le ON le.id = oe.legal_entity_id - JOIN dkb.company_group AS cg ON cg.id = le.company_group_id - LEFT JOIN dkb.recognition_job AS rj + LEFT JOIN {schema}.chart_of_account AS coa ON coa.id = oe.account_id + JOIN {schema}.stored_file AS sf ON sf.id = oe.stored_file_id + JOIN {schema}.legal_entity AS le ON le.id = oe.legal_entity_id + JOIN {schema}.company_group AS cg ON cg.id = le.company_group_id + LEFT JOIN {schema}.recognition_job AS rj ON rj.id = oe.recognition_job_id AND rj.file_id = sf.id WHERE sf.status = 'UPLOADED' AND sf.period_year = p_period_year AND sf.period_quarter = p_period_quarter AND EXISTS ( - SELECT 1 FROM dkb.company_group AS selected_group + SELECT 1 FROM {schema}.company_group AS selected_group WHERE selected_group.id = p_company_group_id ) ) @@ -139,9 +147,10 @@ def upgrade() -> None: def downgrade() -> None: - op.execute("DROP FUNCTION dkb.get_osv_entries_for_group(uuid, integer, text)") - op.execute(""" - CREATE OR REPLACE FUNCTION dkb.get_osv_entries_for_group( + schema = _quoted_db_schema() + op.execute(f"DROP FUNCTION {schema}.get_osv_entries_for_group(uuid, integer, text)") + op.execute(f""" + CREATE OR REPLACE FUNCTION {schema}.get_osv_entries_for_group( p_company_group_id uuid, p_period_year integer, p_period_quarter text @@ -214,8 +223,8 @@ def downgrade() -> None: rj.finished_at, EXISTS ( SELECT 1 - FROM dkb.related_group_member AS member - JOIN dkb.related_group AS rg + FROM {schema}.related_group_member AS member + JOIN {schema}.related_group AS rg ON rg.id = member.related_group_id WHERE member.legal_entity_id = oe.legal_entity_id AND member.is_current IS TRUE @@ -223,26 +232,26 @@ def downgrade() -> None: AND rg.kind = 'GSK' ) AS is_gsk, le.company_group_id = p_company_group_id AS is_gk - FROM dkb.osv_entry AS oe - LEFT JOIN dkb.legal_entity AS counterparty_by_inn + FROM {schema}.osv_entry AS oe + LEFT JOIN {schema}.legal_entity AS counterparty_by_inn ON oe.counterparty_id IS NULL AND counterparty_by_inn.inn = NULLIF(BTRIM(oe.counterparty_inn), '') - LEFT JOIN dkb.legal_entity_alias AS counterparty_by_alias + LEFT JOIN {schema}.legal_entity_alias AS counterparty_by_alias ON oe.counterparty_id IS NULL AND counterparty_by_inn.id IS NULL AND counterparty_by_alias.normalized_alias = NULLIF(BTRIM(oe.counterparty_name), '') - LEFT JOIN dkb.chart_of_account AS coa ON coa.id = oe.account_id - JOIN dkb.stored_file AS sf ON sf.id = oe.stored_file_id - JOIN dkb.legal_entity AS le ON le.id = oe.legal_entity_id - JOIN dkb.company_group AS cg ON cg.id = le.company_group_id - LEFT JOIN dkb.recognition_job AS rj + LEFT JOIN {schema}.chart_of_account AS coa ON coa.id = oe.account_id + JOIN {schema}.stored_file AS sf ON sf.id = oe.stored_file_id + JOIN {schema}.legal_entity AS le ON le.id = oe.legal_entity_id + JOIN {schema}.company_group AS cg ON cg.id = le.company_group_id + LEFT JOIN {schema}.recognition_job AS rj ON rj.id = oe.recognition_job_id AND rj.file_id = sf.id WHERE sf.status = 'UPLOADED' AND sf.period_year = p_period_year AND sf.period_quarter = p_period_quarter AND EXISTS ( - SELECT 1 FROM dkb.company_group AS selected_group + SELECT 1 FROM {schema}.company_group AS selected_group WHERE selected_group.id = p_company_group_id ) ) @@ -256,4 +265,3 @@ def downgrade() -> None: FROM classified_entries AS ce; $function$; """) - diff --git a/api/alembic/versions/d614b414a32f_add_osv_entries_for_group_function.py b/api/alembic/versions/d614b414a32f_add_osv_entries_for_group_function.py index 8c1c46b..e5382a9 100644 --- a/api/alembic/versions/d614b414a32f_add_osv_entries_for_group_function.py +++ b/api/alembic/versions/d614b414a32f_add_osv_entries_for_group_function.py @@ -9,6 +9,7 @@ from typing import Sequence, Union from alembic import op import sqlalchemy as sa +from src.core.config import settings # revision identifiers, used by Alembic. @@ -18,9 +19,16 @@ branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None +def _quoted_db_schema() -> str: + return op.get_bind().dialect.identifier_preparer.quote_schema( + settings.OPENBAO.DATABASE.DATABASE_SCHEMA + ) + + def upgrade() -> None: - op.execute(""" - CREATE OR REPLACE FUNCTION dkb.get_osv_entries_for_group( + schema = _quoted_db_schema() + op.execute(f""" + CREATE OR REPLACE FUNCTION {schema}.get_osv_entries_for_group( p_company_group_id uuid, p_period_year integer, p_period_quarter text @@ -93,8 +101,8 @@ def upgrade() -> None: rj.finished_at, EXISTS ( SELECT 1 - FROM dkb.related_group_member AS member - JOIN dkb.related_group AS rg + FROM {schema}.related_group_member AS member + JOIN {schema}.related_group AS rg ON rg.id = member.related_group_id WHERE member.legal_entity_id = oe.legal_entity_id AND member.is_current IS TRUE @@ -102,19 +110,19 @@ def upgrade() -> None: AND rg.kind = 'GSK' ) AS is_gsk, le.company_group_id = p_company_group_id AS is_gk - FROM dkb.osv_entry AS oe - LEFT JOIN dkb.chart_of_account AS coa ON coa.id = oe.account_id - JOIN dkb.stored_file AS sf ON sf.id = oe.stored_file_id - JOIN dkb.legal_entity AS le ON le.id = oe.legal_entity_id - JOIN dkb.company_group AS cg ON cg.id = le.company_group_id - LEFT JOIN dkb.recognition_job AS rj + FROM {schema}.osv_entry AS oe + LEFT JOIN {schema}.chart_of_account AS coa ON coa.id = oe.account_id + JOIN {schema}.stored_file AS sf ON sf.id = oe.stored_file_id + JOIN {schema}.legal_entity AS le ON le.id = oe.legal_entity_id + JOIN {schema}.company_group AS cg ON cg.id = le.company_group_id + LEFT JOIN {schema}.recognition_job AS rj ON rj.id = oe.recognition_job_id AND rj.file_id = sf.id WHERE sf.status = 'UPLOADED' AND sf.period_year = p_period_year AND sf.period_quarter = p_period_quarter AND EXISTS ( - SELECT 1 FROM dkb.company_group AS selected_group + SELECT 1 FROM {schema}.company_group AS selected_group WHERE selected_group.id = p_company_group_id ) ) @@ -131,7 +139,8 @@ def upgrade() -> None: def downgrade() -> None: - op.execute(""" + schema = _quoted_db_schema() + op.execute(f""" DROP FUNCTION IF EXISTS - dkb.get_osv_entries_for_group(uuid, integer, text) - """) \ No newline at end of file + {schema}.get_osv_entries_for_group(uuid, integer, text) + """) diff --git a/api/src/services/osv_service.py b/api/src/services/osv_service.py index ef5992c..1af1ba2 100644 --- a/api/src/services/osv_service.py +++ b/api/src/services/osv_service.py @@ -1,6 +1,8 @@ import uuid from fastapi import UploadFile +from loguru import logger +from pydantic import ValidationError from sqlalchemy.ext.asyncio import AsyncSession from src.common.exceptions import NotFoundError from src.repositories.osv_repository import OsvEntry, OsvRepository @@ -18,7 +20,27 @@ class OsvService: entries = await self.osv_repo.get_by_company_group( company_group_id=company_group_id, year=year, quarter=quarter.value ) - return [OsvGroupEntryResponse.model_validate(entry) for entry in entries] + + result: list[OsvGroupEntryResponse] = [] + for row_index, entry in enumerate(entries): + try: + result.append(OsvGroupEntryResponse.model_validate(entry)) + except ValidationError as exc: + logger.error( + "Некорректный результат get_osv_entries_for_group: " + "company_group_id={company_group_id}, year={year}, quarter={quarter}, " + "row_index={row_index}, osv_entry_id={osv_entry_id}, " + "validation_errors={validation_errors}", + company_group_id=company_group_id, + year=year, + quarter=quarter.value, + row_index=row_index, + osv_entry_id=entry.get("osv_entry_id"), + validation_errors=exc.errors(include_input=False), + ) + raise + + return result async def get_by_file_id(self, file_id: uuid.UUID):