906 lines
31 KiB
Python
906 lines
31 KiB
Python
"""PostgreSQL snapshot, publication metadata, and ARRIVAL-derived report scope."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import time
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from pathlib import PurePosixPath
|
|
from typing import Any, Callable, List, Optional, Protocol, Sequence, Tuple
|
|
|
|
from monthly_reports.contracts import (
|
|
PROCESSOR_VERSION,
|
|
RESULT_SCHEMA_VERSION,
|
|
ChannelObservation,
|
|
DailyVersionPin,
|
|
ErrorCode,
|
|
MonthlyFact,
|
|
MonthlyReport,
|
|
MonthlySnapshot,
|
|
rule_set_sha256,
|
|
)
|
|
from monthly_reports.core import month_bounds
|
|
|
|
|
|
TARGET_DATABASE = "booking_test"
|
|
DATABASE_ENV = "MONTHLY_REPORT_DATABASE_URL"
|
|
FALLBACK_DATABASE_ENV = "ARR_DATABASE_URL"
|
|
TRANSIENT_SQLSTATES = frozenset({"23505", "40001", "40P01", "55P03"})
|
|
MAX_TRANSACTION_ATTEMPTS = 4
|
|
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
|
|
SAFE_CODE_RE = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$")
|
|
LOCAL_STORAGE_PROVIDER = "local"
|
|
LOCAL_BUCKET_ALIAS = "arr-project-root"
|
|
OSS_STORAGE_PROVIDER = "oss"
|
|
OSS_BUCKET_ALIAS = "arr-private"
|
|
|
|
|
|
SOURCE_FACTS_SQL = """
|
|
SELECT
|
|
facts.id,
|
|
facts.daily_version_id,
|
|
versions.business_date,
|
|
facts.channel_key,
|
|
facts.arrival,
|
|
facts.departure,
|
|
facts.nights,
|
|
facts.adults,
|
|
facts.children,
|
|
facts.block_code,
|
|
facts.no_of_rooms,
|
|
facts.company_name,
|
|
facts.confirmation_no,
|
|
facts.disp_room_no,
|
|
facts.effective_rate_amount,
|
|
facts.full_name,
|
|
facts.res_comment,
|
|
facts.trace_text,
|
|
facts.products,
|
|
facts.rate_code,
|
|
facts.room_category_label,
|
|
facts.real_price,
|
|
facts.total_price,
|
|
facts.kb_amount
|
|
FROM finance.v_active_daily_facts AS facts
|
|
JOIN finance.daily_versions AS versions
|
|
ON versions.id = facts.daily_version_id
|
|
WHERE facts.arrival BETWEEN %s AND %s
|
|
ORDER BY facts.id
|
|
""".strip()
|
|
|
|
CURRENT_DAILY_SQL = """
|
|
SELECT business_date, daily_version_id
|
|
FROM finance.current_daily_versions
|
|
WHERE business_date BETWEEN %s AND %s
|
|
ORDER BY business_date
|
|
""".strip()
|
|
|
|
CHANNEL_OBSERVATIONS_SQL = """
|
|
SELECT
|
|
current_version.business_date,
|
|
current_version.daily_version_id,
|
|
metrics.channel_key,
|
|
metrics.channel_order,
|
|
metrics.row_count
|
|
FROM finance.current_daily_versions AS current_version
|
|
JOIN finance.daily_channel_metrics AS metrics
|
|
ON metrics.daily_version_id = current_version.daily_version_id
|
|
WHERE current_version.business_date BETWEEN %s AND %s
|
|
ORDER BY
|
|
current_version.business_date,
|
|
metrics.channel_order,
|
|
metrics.channel_key
|
|
""".strip()
|
|
|
|
PREFERRED_CHANNELS_SQL = """
|
|
SELECT metrics.channel_key
|
|
FROM finance.current_daily_versions AS current_version
|
|
JOIN finance.daily_channel_metrics AS metrics
|
|
ON metrics.daily_version_id = current_version.daily_version_id
|
|
WHERE current_version.business_date BETWEEN %s AND %s
|
|
GROUP BY metrics.channel_key
|
|
ORDER BY
|
|
min(current_version.business_date),
|
|
min(metrics.channel_order),
|
|
metrics.channel_key
|
|
""".strip()
|
|
|
|
TRIGGER_ARRIVAL_SCOPE_SQL = """
|
|
SELECT min(record.arrival), max(record.arrival)
|
|
FROM finance.daily_versions AS version
|
|
JOIN ingestion.processing_runs AS run
|
|
ON run.id = version.processing_run_id
|
|
JOIN finance.daily_records AS record
|
|
ON record.daily_version_id = version.id
|
|
WHERE version.id = %s
|
|
AND run.run_status = 'accepted'
|
|
AND record.outcome = 'retained'
|
|
""".strip()
|
|
|
|
MONTH_MAX_ARRIVAL_SQL = """
|
|
SELECT max(fact.arrival)
|
|
FROM finance.v_active_daily_facts AS fact
|
|
WHERE fact.arrival >= %s
|
|
AND fact.arrival < %s
|
|
""".strip()
|
|
|
|
|
|
class RepositoryError(RuntimeError):
|
|
def __init__(self, code: str, safe_message: str):
|
|
super().__init__(safe_message)
|
|
self.code = code
|
|
self.safe_message = safe_message
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DatabaseConfig:
|
|
dsn: str
|
|
|
|
@classmethod
|
|
def from_environment(cls) -> "DatabaseConfig":
|
|
dsn = os.environ.get(DATABASE_ENV, "").strip()
|
|
if not dsn:
|
|
dsn = os.environ.get(FALLBACK_DATABASE_ENV, "").strip()
|
|
if not dsn:
|
|
raise RepositoryError(
|
|
ErrorCode.REQUEST_INVALID,
|
|
f"{DATABASE_ENV} or {FALLBACK_DATABASE_ENV} is required",
|
|
)
|
|
return cls(dsn=dsn)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FileMetadata:
|
|
file_kind: str
|
|
original_filename: str
|
|
storage_key: str
|
|
sha256: str
|
|
byte_size: int
|
|
mime_type: str
|
|
storage_provider: str = LOCAL_STORAGE_PROVIDER
|
|
bucket_alias: str = LOCAL_BUCKET_ALIAS
|
|
|
|
def validate(self) -> None:
|
|
storage = PurePosixPath(self.storage_key)
|
|
if (
|
|
self.file_kind not in {"monthly_xlsx", "result_json"}
|
|
or not self.storage_key
|
|
or storage.is_absolute()
|
|
or ".." in storage.parts
|
|
or "\\" in self.storage_key
|
|
or "\x00" in self.storage_key
|
|
or self.original_filename != PurePosixPath(self.original_filename).name
|
|
or "/" in self.original_filename
|
|
or not SHA256_RE.fullmatch(self.sha256)
|
|
or self.byte_size <= 0
|
|
or not self.mime_type.strip()
|
|
or not isinstance(self.storage_provider, str)
|
|
or self.storage_provider not in {"local", "local_fixture", "oss", "s3"}
|
|
or not isinstance(self.bucket_alias, str)
|
|
or not self.bucket_alias.strip()
|
|
or "/" in self.bucket_alias
|
|
or "\\" in self.bucket_alias
|
|
):
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"generated artifact metadata is invalid",
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReservedReport:
|
|
report_version_id: int
|
|
version_no: int
|
|
period_start: Optional[date] = None
|
|
as_of_date: Optional[date] = None
|
|
daily_versions: Tuple[DailyVersionPin, ...] = tuple()
|
|
source_snapshot_sha256: Optional[str] = None
|
|
report_status: str = "reserved"
|
|
published_artifact: Optional[FileMetadata] = None
|
|
|
|
@property
|
|
def already_published(self) -> bool:
|
|
return (
|
|
self.report_status in {"active", "superseded"}
|
|
and self.published_artifact is not None
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DerivedMonthlyRequest:
|
|
report_year: int
|
|
report_month: int
|
|
as_of_date: date
|
|
|
|
|
|
class ReportRepository(Protocol):
|
|
def load_snapshot(self, year: int, month: int, as_of_date: date) -> MonthlySnapshot:
|
|
...
|
|
|
|
def reserve_report(self, report: MonthlyReport) -> ReservedReport:
|
|
...
|
|
|
|
def activate_report(
|
|
self,
|
|
reservation: ReservedReport,
|
|
artifact: FileMetadata,
|
|
result_json: FileMetadata,
|
|
semantic_sha256: str,
|
|
) -> None:
|
|
...
|
|
|
|
def mark_failed(self, reservation: ReservedReport, code: str, safe_message: str) -> None:
|
|
...
|
|
|
|
|
|
def _default_connect(dsn: str) -> Any:
|
|
try:
|
|
import psycopg # type: ignore[import-not-found]
|
|
except ImportError:
|
|
raise RepositoryError(
|
|
ErrorCode.INTERNAL_ERROR,
|
|
"PostgreSQL driver is unavailable; install requirements-monthly-reports.txt",
|
|
) from None
|
|
try:
|
|
return psycopg.connect(dsn, autocommit=False)
|
|
except Exception:
|
|
raise RepositoryError(
|
|
ErrorCode.DATABASE_FAILED,
|
|
"database connection failed",
|
|
) from None
|
|
|
|
|
|
def _decimal(value: Any) -> Decimal:
|
|
return value if isinstance(value, Decimal) else Decimal(str(value))
|
|
|
|
|
|
def _is_transient_database_error(error: Exception) -> bool:
|
|
sqlstate = getattr(error, "sqlstate", None)
|
|
if not isinstance(sqlstate, str):
|
|
sqlstate = getattr(getattr(error, "diag", None), "sqlstate", None)
|
|
return isinstance(sqlstate, str) and sqlstate in TRANSIENT_SQLSTATES
|
|
|
|
|
|
class PostgresReportRepository:
|
|
"""Loads one repeatable snapshot and atomically publishes its lineage."""
|
|
|
|
def __init__(
|
|
self,
|
|
config: DatabaseConfig,
|
|
connect: Optional[Callable[[str], Any]] = None,
|
|
) -> None:
|
|
self._config = config
|
|
self._connect = connect or _default_connect
|
|
|
|
def _open(self) -> Any:
|
|
try:
|
|
return self._connect(self._config.dsn)
|
|
except RepositoryError:
|
|
raise
|
|
except Exception:
|
|
raise RepositoryError(
|
|
ErrorCode.DATABASE_FAILED,
|
|
"database connection failed",
|
|
) from None
|
|
|
|
@staticmethod
|
|
def _assert_target(cursor: Any) -> None:
|
|
cursor.execute("SELECT current_database()")
|
|
row = cursor.fetchone()
|
|
if not row or row[0] != TARGET_DATABASE:
|
|
raise RepositoryError(
|
|
ErrorCode.DATABASE_FAILED,
|
|
"database target is not booking_test",
|
|
)
|
|
|
|
@classmethod
|
|
def _begin_serializable(cls, cursor: Any, *, read_only: bool = False) -> None:
|
|
cursor.execute(
|
|
"SET TRANSACTION ISOLATION LEVEL SERIALIZABLE READ ONLY"
|
|
if read_only
|
|
else "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE"
|
|
)
|
|
cursor.execute("SET LOCAL lock_timeout = '10s'")
|
|
cursor.execute("SET LOCAL statement_timeout = '30s'")
|
|
cls._assert_target(cursor)
|
|
|
|
def _run_serializable(
|
|
self,
|
|
operation: Callable[[Any], Any],
|
|
safe_message: str,
|
|
*,
|
|
read_only: bool = False,
|
|
) -> Any:
|
|
for attempt_index in range(MAX_TRANSACTION_ATTEMPTS):
|
|
connection = self._open()
|
|
retry = False
|
|
try:
|
|
with connection.transaction():
|
|
with connection.cursor() as cursor:
|
|
self._begin_serializable(cursor, read_only=read_only)
|
|
return operation(cursor)
|
|
except RepositoryError:
|
|
raise
|
|
except Exception as error:
|
|
retry = (
|
|
attempt_index + 1 < MAX_TRANSACTION_ATTEMPTS
|
|
and _is_transient_database_error(error)
|
|
)
|
|
if not retry:
|
|
raise RepositoryError(ErrorCode.DATABASE_FAILED, safe_message) from None
|
|
finally:
|
|
connection.close()
|
|
if retry:
|
|
time.sleep(0.02 * (2**attempt_index))
|
|
raise RepositoryError(ErrorCode.DATABASE_FAILED, safe_message)
|
|
|
|
def load_snapshot(self, year: int, month: int, as_of_date: date) -> MonthlySnapshot:
|
|
month_start, _month_end = month_bounds(year, month)
|
|
connection = self._open()
|
|
try:
|
|
with connection.transaction():
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY"
|
|
)
|
|
cursor.execute("SET LOCAL statement_timeout = '30s'")
|
|
self._assert_target(cursor)
|
|
cursor.execute(SOURCE_FACTS_SQL, (month_start, as_of_date))
|
|
fact_rows = list(cursor.fetchall())
|
|
cursor.execute(CURRENT_DAILY_SQL, (month_start, as_of_date))
|
|
pin_rows = list(cursor.fetchall())
|
|
cursor.execute(CHANNEL_OBSERVATIONS_SQL, (month_start, as_of_date))
|
|
observation_rows = list(cursor.fetchall())
|
|
cursor.execute(PREFERRED_CHANNELS_SQL, (month_start, _month_end))
|
|
preferred_rows = list(cursor.fetchall())
|
|
|
|
facts = tuple(
|
|
MonthlyFact(
|
|
daily_record_id=int(row[0]),
|
|
daily_version_id=int(row[1]),
|
|
business_date=row[2],
|
|
channel_key=str(row[3] or ""),
|
|
arrival=row[4],
|
|
departure=row[5],
|
|
nights=int(row[6]),
|
|
adults=int(row[7]),
|
|
children=int(row[8]),
|
|
block_code=row[9],
|
|
no_of_rooms=int(row[10]),
|
|
company_name=str(row[11] or ""),
|
|
confirmation_no=str(row[12] or ""),
|
|
disp_room_no=str(row[13] or ""),
|
|
effective_rate_amount=_decimal(row[14]),
|
|
full_name=str(row[15] or ""),
|
|
res_comment=row[16],
|
|
trace_text=row[17],
|
|
products=row[18],
|
|
rate_code=str(row[19] or ""),
|
|
room_category_label=row[20],
|
|
real_price=_decimal(row[21]),
|
|
total_price=_decimal(row[22]),
|
|
kb_amount=None if row[23] is None else _decimal(row[23]),
|
|
)
|
|
for row in fact_rows
|
|
)
|
|
pins = tuple(
|
|
DailyVersionPin(business_date=row[0], daily_version_id=int(row[1]))
|
|
for row in pin_rows
|
|
)
|
|
observations = tuple(
|
|
ChannelObservation(
|
|
business_date=row[0],
|
|
daily_version_id=int(row[1]),
|
|
worksheet=str(row[2] or ""),
|
|
worksheet_order=None if row[3] is None else int(row[3]),
|
|
row_count=int(row[4]),
|
|
)
|
|
for row in observation_rows
|
|
)
|
|
return MonthlySnapshot(
|
|
facts=facts,
|
|
daily_versions=pins,
|
|
channel_observations=observations,
|
|
preferred_channel_order=tuple(str(row[0] or "") for row in preferred_rows),
|
|
)
|
|
except RepositoryError:
|
|
raise
|
|
except Exception:
|
|
raise RepositoryError(
|
|
ErrorCode.DATABASE_FAILED,
|
|
"database snapshot query failed",
|
|
) from None
|
|
finally:
|
|
connection.close()
|
|
|
|
def derive_monthly_request(
|
|
self,
|
|
daily_version_id: int,
|
|
) -> Optional[DerivedMonthlyRequest]:
|
|
if isinstance(daily_version_id, bool) or daily_version_id < 1:
|
|
raise RepositoryError(
|
|
ErrorCode.REQUEST_INVALID,
|
|
"daily version identity is invalid",
|
|
)
|
|
|
|
def operation(cursor: Any) -> Optional[DerivedMonthlyRequest]:
|
|
cursor.execute(TRIGGER_ARRIVAL_SCOPE_SQL, (daily_version_id,))
|
|
row = cursor.fetchone()
|
|
if not row or row[0] is None or row[1] is None:
|
|
return None
|
|
first_arrival, last_arrival = row[0], row[1]
|
|
if (
|
|
first_arrival.year != last_arrival.year
|
|
or first_arrival.month != last_arrival.month
|
|
):
|
|
raise RepositoryError(
|
|
ErrorCode.SOURCE_INVALID,
|
|
"one daily version spans multiple ARRIVAL months",
|
|
)
|
|
period_start, period_end = month_bounds(
|
|
first_arrival.year,
|
|
first_arrival.month,
|
|
)
|
|
cursor.execute(
|
|
MONTH_MAX_ARRIVAL_SQL,
|
|
(period_start, period_end + timedelta(days=1)),
|
|
)
|
|
latest = cursor.fetchone()
|
|
as_of_date = latest[0] if latest else None
|
|
if as_of_date is None:
|
|
return None
|
|
return DerivedMonthlyRequest(
|
|
report_year=as_of_date.year,
|
|
report_month=as_of_date.month,
|
|
as_of_date=as_of_date,
|
|
)
|
|
|
|
return self._run_serializable(
|
|
operation,
|
|
"monthly ARRIVAL scope query failed",
|
|
read_only=True,
|
|
)
|
|
|
|
@staticmethod
|
|
def _lock_scope(cursor: Any, period_start: date, period_end: date) -> None:
|
|
lock_key = f"monthly_channel:{period_start.isoformat()}:{period_end.isoformat()}:*"
|
|
cursor.execute(
|
|
"SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
|
|
(lock_key,),
|
|
)
|
|
|
|
@staticmethod
|
|
def _current_pins(
|
|
cursor: Any,
|
|
period_start: date,
|
|
as_of_date: date,
|
|
) -> Tuple[DailyVersionPin, ...]:
|
|
cursor.execute(
|
|
CURRENT_DAILY_SQL,
|
|
(period_start, as_of_date),
|
|
)
|
|
return tuple(
|
|
DailyVersionPin(business_date=row[0], daily_version_id=int(row[1]))
|
|
for row in cursor.fetchall()
|
|
)
|
|
|
|
@classmethod
|
|
def _validate_pins_current(cls, cursor: Any, report: MonthlyReport) -> None:
|
|
period_start, _period_end = month_bounds(report.report_year, report.report_month)
|
|
if cls._current_pins(cursor, period_start, report.as_of_date) != report.daily_versions:
|
|
raise RepositoryError(
|
|
ErrorCode.SOURCE_SNAPSHOT_STALE,
|
|
"daily source versions changed while the monthly report was being generated",
|
|
)
|
|
|
|
@staticmethod
|
|
def _source_snapshot_sha256(report: MonthlyReport) -> str:
|
|
identity = {
|
|
"period_start": date(
|
|
report.report_year,
|
|
report.report_month,
|
|
1,
|
|
).isoformat(),
|
|
"as_of_date": report.as_of_date.isoformat(),
|
|
"processor_version": PROCESSOR_VERSION,
|
|
"rule_set_sha256": rule_set_sha256(),
|
|
"result_schema_version": RESULT_SCHEMA_VERSION,
|
|
"daily_versions": [
|
|
[pin.business_date.isoformat(), pin.daily_version_id]
|
|
for pin in report.daily_versions
|
|
],
|
|
"channel_manifest": [list(item) for item in report.channel_manifest],
|
|
}
|
|
return hashlib.sha256(
|
|
json.dumps(
|
|
identity,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
@staticmethod
|
|
def _published_artifact(row: Sequence[Any]) -> Optional[FileMetadata]:
|
|
if row[3] is None:
|
|
return None
|
|
if len(row) >= 11:
|
|
provider = str(row[4] or LOCAL_STORAGE_PROVIDER)
|
|
bucket_alias = str(row[5] or LOCAL_BUCKET_ALIAS)
|
|
filename_index, key_index, hash_index, size_index, mime_index = 6, 7, 8, 9, 10
|
|
else:
|
|
provider = LOCAL_STORAGE_PROVIDER
|
|
bucket_alias = LOCAL_BUCKET_ALIAS
|
|
filename_index, key_index, hash_index, size_index, mime_index = 4, 5, 6, 7, 8
|
|
metadata = FileMetadata(
|
|
file_kind=str(row[3]),
|
|
original_filename=str(row[filename_index]),
|
|
storage_key=str(row[key_index]),
|
|
sha256=str(row[hash_index]),
|
|
byte_size=int(row[size_index]),
|
|
mime_type=str(row[mime_index] or "application/octet-stream"),
|
|
storage_provider=provider,
|
|
bucket_alias=bucket_alias,
|
|
)
|
|
metadata.validate()
|
|
return metadata
|
|
|
|
def reserve_report(self, report: MonthlyReport) -> ReservedReport:
|
|
period_start, period_end = month_bounds(report.report_year, report.report_month)
|
|
snapshot_sha256 = self._source_snapshot_sha256(report)
|
|
|
|
def operation(cursor: Any) -> ReservedReport:
|
|
self._lock_scope(cursor, period_start, period_end)
|
|
self._validate_pins_current(cursor, report)
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
run.id,
|
|
run.version_no,
|
|
run.report_status,
|
|
artifact.artifact_kind,
|
|
artifact.storage_provider,
|
|
artifact.bucket_alias,
|
|
artifact.original_filename,
|
|
artifact.object_key,
|
|
artifact.sha256,
|
|
artifact.byte_size,
|
|
artifact.mime_type
|
|
FROM reporting.monthly_runs AS run
|
|
LEFT JOIN ingestion.artifacts AS artifact
|
|
ON artifact.id = run.workbook_artifact_id
|
|
WHERE run.period_start = %s
|
|
AND run.source_snapshot_sha256 = %s
|
|
FOR UPDATE OF run
|
|
""",
|
|
(period_start, snapshot_sha256),
|
|
)
|
|
existing = cursor.fetchone()
|
|
if existing:
|
|
status = str(existing[2])
|
|
if status == "failed":
|
|
cursor.execute(
|
|
"""
|
|
UPDATE reporting.monthly_runs
|
|
SET report_status = 'reserved',
|
|
failure_code = NULL,
|
|
failure_message = NULL,
|
|
failed_at = NULL,
|
|
updated_at = now()
|
|
WHERE id = %s
|
|
""",
|
|
(int(existing[0]),),
|
|
)
|
|
status = "reserved"
|
|
return ReservedReport(
|
|
report_version_id=int(existing[0]),
|
|
version_no=int(existing[1]),
|
|
period_start=period_start,
|
|
as_of_date=report.as_of_date,
|
|
daily_versions=report.daily_versions,
|
|
source_snapshot_sha256=snapshot_sha256,
|
|
report_status=status,
|
|
published_artifact=self._published_artifact(existing),
|
|
)
|
|
|
|
cursor.execute(
|
|
"""
|
|
SELECT COALESCE(max(version_no), 0) + 1
|
|
FROM reporting.monthly_runs
|
|
WHERE period_start = %s
|
|
""",
|
|
(period_start,),
|
|
)
|
|
version_no = int(cursor.fetchone()[0])
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO reporting.monthly_runs (
|
|
period_start,
|
|
as_of_date,
|
|
version_no,
|
|
source_snapshot_sha256,
|
|
report_status,
|
|
processor_version,
|
|
rule_set_sha256,
|
|
result_schema_version,
|
|
row_count,
|
|
channel_count
|
|
)
|
|
VALUES (%s, %s, %s, %s, 'reserved', %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
period_start,
|
|
report.as_of_date,
|
|
version_no,
|
|
snapshot_sha256,
|
|
PROCESSOR_VERSION,
|
|
rule_set_sha256(),
|
|
RESULT_SCHEMA_VERSION,
|
|
report.row_count,
|
|
len(report.channel_manifest),
|
|
),
|
|
)
|
|
report_id = int(cursor.fetchone()[0])
|
|
for pin in report.daily_versions:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO reporting.monthly_run_daily_versions (
|
|
report_id, business_date, daily_version_id
|
|
)
|
|
VALUES (%s, %s, %s)
|
|
""",
|
|
(report_id, pin.business_date, pin.daily_version_id),
|
|
)
|
|
for worksheet, worksheet_order, row_count in report.channel_manifest:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO reporting.monthly_channel_manifest (
|
|
report_id, worksheet, worksheet_order, row_count
|
|
)
|
|
VALUES (%s, %s, %s, %s)
|
|
""",
|
|
(report_id, worksheet, worksheet_order, row_count),
|
|
)
|
|
return ReservedReport(
|
|
report_version_id=report_id,
|
|
version_no=version_no,
|
|
period_start=period_start,
|
|
as_of_date=report.as_of_date,
|
|
daily_versions=report.daily_versions,
|
|
source_snapshot_sha256=snapshot_sha256,
|
|
)
|
|
|
|
return self._run_serializable(
|
|
operation,
|
|
"monthly report source reservation failed",
|
|
)
|
|
|
|
@staticmethod
|
|
def _ensure_artifact(cursor: Any, metadata: FileMetadata) -> int:
|
|
cursor.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
artifact_kind,
|
|
original_filename,
|
|
sha256,
|
|
byte_size,
|
|
mime_type
|
|
FROM ingestion.artifacts
|
|
WHERE storage_provider = %s
|
|
AND bucket_alias = %s
|
|
AND object_key = %s
|
|
AND object_version_id IS NULL
|
|
FOR SHARE
|
|
""",
|
|
(metadata.storage_provider, metadata.bucket_alias, metadata.storage_key),
|
|
)
|
|
row = cursor.fetchone()
|
|
if row:
|
|
if (
|
|
str(row[1]) != metadata.file_kind
|
|
or str(row[2]) != metadata.original_filename
|
|
or str(row[3]) != metadata.sha256
|
|
or int(row[4]) != metadata.byte_size
|
|
or str(row[5] or "") != metadata.mime_type
|
|
):
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"stored monthly artifact identity conflicts",
|
|
)
|
|
return int(row[0])
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO ingestion.artifacts (
|
|
artifact_kind,
|
|
storage_provider,
|
|
bucket_alias,
|
|
object_key,
|
|
original_filename,
|
|
sha256,
|
|
byte_size,
|
|
mime_type
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
metadata.file_kind,
|
|
metadata.storage_provider,
|
|
metadata.bucket_alias,
|
|
metadata.storage_key,
|
|
metadata.original_filename,
|
|
metadata.sha256,
|
|
metadata.byte_size,
|
|
metadata.mime_type,
|
|
),
|
|
)
|
|
return int(cursor.fetchone()[0])
|
|
|
|
def activate_report(
|
|
self,
|
|
reservation: ReservedReport,
|
|
artifact: FileMetadata,
|
|
result_json: FileMetadata,
|
|
semantic_sha256: str,
|
|
) -> None:
|
|
artifact.validate()
|
|
result_json.validate()
|
|
if not SHA256_RE.fullmatch(semantic_sha256):
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"monthly semantic identity is invalid",
|
|
)
|
|
if reservation.period_start is None or reservation.as_of_date is None:
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"monthly source reservation is incomplete",
|
|
)
|
|
|
|
def operation(cursor: Any) -> None:
|
|
period_start, period_end = month_bounds(
|
|
reservation.period_start.year,
|
|
reservation.period_start.month,
|
|
)
|
|
self._lock_scope(cursor, period_start, period_end)
|
|
current = self._current_pins(
|
|
cursor,
|
|
period_start,
|
|
reservation.as_of_date,
|
|
)
|
|
if current != reservation.daily_versions:
|
|
raise RepositoryError(
|
|
ErrorCode.SOURCE_SNAPSHOT_STALE,
|
|
"daily source versions changed before publication",
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
SELECT report_status, source_snapshot_sha256
|
|
FROM reporting.monthly_runs
|
|
WHERE id = %s
|
|
AND period_start = %s
|
|
AND version_no = %s
|
|
FOR UPDATE
|
|
""",
|
|
(
|
|
reservation.report_version_id,
|
|
period_start,
|
|
reservation.version_no,
|
|
),
|
|
)
|
|
run = cursor.fetchone()
|
|
if (
|
|
not run
|
|
or reservation.source_snapshot_sha256 is None
|
|
or str(run[1]) != reservation.source_snapshot_sha256
|
|
):
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"monthly report reservation identity changed",
|
|
)
|
|
workbook_artifact_id = self._ensure_artifact(cursor, artifact)
|
|
result_artifact_id = self._ensure_artifact(cursor, result_json)
|
|
if str(run[0]) in {"active", "superseded"}:
|
|
cursor.execute(
|
|
"""
|
|
SELECT workbook_artifact_id, result_artifact_id, semantic_sha256
|
|
FROM reporting.monthly_runs
|
|
WHERE id = %s
|
|
""",
|
|
(reservation.report_version_id,),
|
|
)
|
|
published = cursor.fetchone()
|
|
if (
|
|
not published
|
|
or int(published[0]) != workbook_artifact_id
|
|
or int(published[1]) != result_artifact_id
|
|
or str(published[2]) != semantic_sha256
|
|
):
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"published monthly report identity conflicts",
|
|
)
|
|
return
|
|
if str(run[0]) != "reserved":
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"monthly report reservation is not publishable",
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
UPDATE reporting.monthly_runs
|
|
SET report_status = 'superseded',
|
|
superseded_at = now(),
|
|
updated_at = now()
|
|
WHERE period_start = %s
|
|
AND report_status = 'active'
|
|
AND id <> %s
|
|
""",
|
|
(period_start, reservation.report_version_id),
|
|
)
|
|
cursor.execute(
|
|
"""
|
|
UPDATE reporting.monthly_runs
|
|
SET report_status = 'active',
|
|
workbook_artifact_id = %s,
|
|
result_artifact_id = %s,
|
|
semantic_sha256 = %s,
|
|
published_at = now(),
|
|
updated_at = now()
|
|
WHERE id = %s
|
|
AND report_status = 'reserved'
|
|
""",
|
|
(
|
|
workbook_artifact_id,
|
|
result_artifact_id,
|
|
semantic_sha256,
|
|
reservation.report_version_id,
|
|
),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
raise RepositoryError(
|
|
ErrorCode.PUBLISH_FAILED,
|
|
"monthly report activation did not complete",
|
|
)
|
|
|
|
self._run_serializable(
|
|
operation,
|
|
"monthly report source revalidation failed",
|
|
)
|
|
|
|
def mark_failed(
|
|
self,
|
|
reservation: ReservedReport,
|
|
code: str,
|
|
safe_message: str,
|
|
) -> None:
|
|
if not SAFE_CODE_RE.fullmatch(code):
|
|
code = ErrorCode.INTERNAL_ERROR
|
|
message = str(safe_message).strip()[:500] or "monthly report processing failed"
|
|
|
|
def operation(cursor: Any) -> None:
|
|
cursor.execute(
|
|
"""
|
|
UPDATE reporting.monthly_runs
|
|
SET report_status = 'failed',
|
|
failure_code = %s,
|
|
failure_message = %s,
|
|
failed_at = now(),
|
|
updated_at = now()
|
|
WHERE id = %s
|
|
AND report_status = 'reserved'
|
|
""",
|
|
(code, message, reservation.report_version_id),
|
|
)
|
|
|
|
self._run_serializable(
|
|
operation,
|
|
"monthly report failure state could not be recorded",
|
|
)
|