Wire the one-off Calico S3 load runner

run_calico_s3_load reads the Calico export from S3 and runs it through
run_calico_load against portfolio 824, resolving references to UPRNs via the
property table and persisting with ConditionPostgres. S3 + Postgres wiring
around the unit-tested load; run where a database and S3 are available.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-07-13 15:15:38 +00:00
parent 773869ada4
commit 3a77a1a811

View file

@ -0,0 +1,56 @@
"""One-off runner for the Calico stock-condition (roof-covering) load.
Calico is a bulk one-off import, not a recurring feed, so it runs through this
runner rather than the SQS-Lambda path Peabody/LBWF use (ADR-0064). The raw
export lives in S3; references resolve to UPRNs via the ``property`` table for
Calico's portfolio.
Not exercised by the unit suite -- it is S3 + Postgres wiring around the
unit-tested ``run_calico_load``; run it where a database and S3 are available.
"""
import os
from io import BytesIO
from applications.condition.calico_load import run_calico_load
from applications.condition.load_report import LoadReport
from infrastructure.postgres.config import PostgresConfig
from infrastructure.postgres.engine import make_engine, transactional_session
from repositories.condition.condition_postgres import ConditionPostgres
from repositories.condition.property_reference_reader_postgres import (
PropertyReferenceReaderPostgres,
)
from repositories.condition.property_uprn_lookup import PropertyUprnLookup
from utils.logger import setup_logger
from utils.s3 import read_io_from_s3
logger = setup_logger()
CALICO_PORTFOLIO_ID = 824
# The export date the Calico file was generated (Roof Covering 2.6.26); anchors
# renewal years so they don't drift on re-run (ADR-0064).
CALICO_EXPORT_YEAR = 2026
def run_calico_s3_load(
bucket: str,
key: str,
portfolio_id: int = CALICO_PORTFOLIO_ID,
export_year: int = CALICO_EXPORT_YEAR,
) -> LoadReport:
file_bytes: BytesIO = read_io_from_s3(bucket_name=bucket, file_key=key)
engine = make_engine(PostgresConfig.from_env(dict(os.environ)))
with transactional_session(engine) as session:
lookup = PropertyUprnLookup(
PropertyReferenceReaderPostgres(session), portfolio_id
)
report = run_calico_load(
file_stream=file_bytes,
uprn_lookup=lookup,
export_year=export_year,
writer=ConditionPostgres(),
)
logger.info(f"[calico_runner] {report}")
return report