mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
# import json
|
|
from typing import Mapping, Any
|
|
import os
|
|
|
|
# from io import BytesIO
|
|
|
|
# from backend.condition.condition_trigger_request import ConditionTriggerRequest
|
|
# from backend.condition.lookups.uprn_lookup_s3 import UprnLookupS3
|
|
# from backend.condition.processor import process_file
|
|
# from utils.logger import setup_logger
|
|
# from utils.s3 import read_io_from_s3
|
|
|
|
|
|
# logger = setup_logger()
|
|
|
|
|
|
def handler(event: Mapping[str, Any], context: Any) -> None:
|
|
print(
|
|
"hello Jun-te",
|
|
os.getenv("JUNTE", "empty junte"),
|
|
)
|
|
print(
|
|
"hello DEV DB HOST:",
|
|
os.getenv("DEV_DB_HOST", "empty db"),
|
|
)
|
|
print(
|
|
"hello REAL DB HOST:",
|
|
os.getenv("REAL_DB_HOST", "empty db"),
|
|
)
|
|
print(
|
|
"hello access key",
|
|
os.getenv("AWS_ACCESS_KEY_ID", "empty key"),
|
|
)
|
|
print(
|
|
"hello region",
|
|
os.getenv("AWS_REGION", "empty region"),
|
|
)
|
|
# uprn_lookup = UprnLookupS3(
|
|
# bucket="", key=""
|
|
# ) # TODO: replace with postgres implementation
|
|
|
|
# for record in event.get("Records", []):
|
|
# try:
|
|
# body_dict = json.loads(record["body"])
|
|
# payload = ConditionTriggerRequest.model_validate(body_dict)
|
|
|
|
# file_bytes: BytesIO = read_io_from_s3(
|
|
# bucket_name=payload.trigger_file_bucket,
|
|
# file_key=payload.trigger_file_key,
|
|
# )
|
|
|
|
# process_file(
|
|
# file_stream=file_bytes,
|
|
# file_type=payload.file_type,
|
|
# uprn_lookup=uprn_lookup,
|
|
# )
|
|
|
|
# except Exception as e:
|
|
# logger.error(f"Failed to process record: {e}")
|