Add CI workflow and apply ruff formatting (#13)

This commit is contained in:
Adam Moussa 2026-05-08 15:49:07 -04:00 • committed by GitHub
parent 723753d004
commit 5ff6f87c6f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 57 additions and 14 deletions

10
.github/workflows/ci.yaml vendored Normal file
View file

@ -0,0 +1,10 @@
name: CI
on:
pull_request:
branches: [main]
jobs:
ci:
uses: Sea-Haven-Industries/.github/.github/workflows/ci-python-sam.yaml@main
with:
source-dirs: "src"

View file

@ -14,6 +14,7 @@ logger.setLevel(logging.INFO)
# US Eastern (handles DST automatically via zoneinfo on Python 3.9+) # US Eastern (handles DST automatically via zoneinfo on Python 3.9+)
try: try:
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
EASTERN = ZoneInfo("America/New_York") EASTERN = ZoneInfo("America/New_York")
except ImportError: except ImportError:
EASTERN = timezone(timedelta(hours=-5)) EASTERN = timezone(timedelta(hours=-5))
@ -21,7 +22,9 @@ except ImportError:
def load_schedule() -> dict: def load_schedule() -> dict:
"""Load schedule from config bundled in the Lambda package.""" """Load schedule from config bundled in the Lambda package."""
config_path = os.path.join(os.path.dirname(__file__), "..", "config", "schedule.yaml") config_path = os.path.join(
os.path.dirname(__file__), "..", "config", "schedule.yaml"
)
with open(config_path) as f: with open(config_path) as f:
return yaml.safe_load(f) return yaml.safe_load(f)
@ -44,7 +47,9 @@ def get_3cx_credentials() -> dict:
WEEKEND_DAYS = {"Saturday", "Sunday"} WEEKEND_DAYS = {"Saturday", "Sunday"}
def check_dynamodb_override(date_str: str, day_name: str, shift_type: str = "night") -> str | None: def check_dynamodb_override(
date_str: str, day_name: str, shift_type: str = "night"
) -> str | None:
"""Check the afterhours-shifts DynamoDB table for a Slack-driven override. """Check the afterhours-shifts DynamoDB table for a Slack-driven override.
Args: Args:
@ -72,9 +77,16 @@ def check_dynamodb_override(date_str: str, day_name: str, shift_type: str = "nig
# Open shift with no pickup — use fallback from DynamoDB config or default 100 # Open shift with no pickup — use fallback from DynamoDB config or default 100
config_resp = table.get_item(Key={"PK": "CONFIG", "SK": "CONFIG"}) config_resp = table.get_item(Key={"PK": "CONFIG", "SK": "CONFIG"})
fallback = config_resp.get("Item", {}).get("fallback_extension", "100") fallback = config_resp.get("Item", {}).get("fallback_extension", "100")
logger.info("DynamoDB override for %s (%s): OPEN — using fallback Ext %s", override_sk, shift_type, fallback) logger.info(
"DynamoDB override for %s (%s): OPEN — using fallback Ext %s",
override_sk,
shift_type,
fallback,
)
return fallback return fallback
logger.info("DynamoDB override for %s (%s): Ext %s", override_sk, shift_type, ext) logger.info(
"DynamoDB override for %s (%s): Ext %s", override_sk, shift_type, ext
)
return ext return ext
# Check DynamoDB weekly schedule (takes precedence over YAML when SHIFT_TABLE is set) # Check DynamoDB weekly schedule (takes precedence over YAML when SHIFT_TABLE is set)
@ -83,7 +95,13 @@ def check_dynamodb_override(date_str: str, day_name: str, shift_type: str = "nig
item = resp.get("Item") item = resp.get("Item")
if item: if item:
ext = item["extension"] ext = item["extension"]
logger.info("DynamoDB weekly schedule for %s (%s, %s): Ext %s", date_str, weekly_sk, shift_type, ext) logger.info(
"DynamoDB weekly schedule for %s (%s, %s): Ext %s",
date_str,
weekly_sk,
shift_type,
ext,
)
return ext return ext
return None return None
@ -126,7 +144,9 @@ def resolve_extension(schedule: dict, now: datetime, shift_type: str = "night")
# YAML config only covers night shifts (legacy) # YAML config only covers night shifts (legacy)
if shift_type == "day": if shift_type == "day":
logger.info("No DynamoDB entry for %s day shift — using fallback Ext 100", date_str) logger.info(
"No DynamoDB entry for %s day shift — using fallback Ext 100", date_str
)
return "100" return "100"
# Fall back to YAML config # Fall back to YAML config
@ -140,7 +160,9 @@ def resolve_extension(schedule: dict, now: datetime, shift_type: str = "night")
if day_name not in weekly: if day_name not in weekly:
raise ValueError(f"No schedule entry for {day_name}") raise ValueError(f"No schedule entry for {day_name}")
ext = str(weekly[day_name]) ext = str(weekly[day_name])
logger.info("Using YAML weekly schedule for %s (%s): Ext %s", date_str, day_name, ext) logger.info(
"Using YAML weekly schedule for %s (%s): Ext %s", date_str, day_name, ext
)
return ext return ext
@ -158,8 +180,15 @@ def handler(event, context):
force = event.get("force", False) force = event.get("force", False)
valid_hours = {8, 17} if day_name in WEEKEND_DAYS else {8} valid_hours = {8, 17} if day_name in WEEKEND_DAYS else {8}
if not force and current_hour not in valid_hours: if not force and current_hour not in valid_hours:
logger.info("Current ET hour is %d on %s — skipping (wrong DST rule fired)", current_hour, day_name) logger.info(
return {"skipped": True, "reason": f"ET hour is {current_hour}, not in {valid_hours}"} "Current ET hour is %d on %s — skipping (wrong DST rule fired)",
current_hour,
day_name,
)
return {
"skipped": True,
"reason": f"ET hour is {current_hour}, not in {valid_hours}",
}
schedule = load_schedule() schedule = load_schedule()
shift_type = determine_shift_type(now) shift_type = determine_shift_type(now)

View file

@ -16,13 +16,17 @@ class ThreeCXClient:
""" """
self.base_url = f"https://{domain}" self.base_url = f"https://{domain}"
self.session = requests.Session() self.session = requests.Session()
self.session.headers.update({ self.session.headers.update(
"OData-Version": "4.0", {
"Content-Type": "application/json", "OData-Version": "4.0",
}) "Content-Type": "application/json",
}
)
if auth_mode == "oauth": if auth_mode == "oauth":
self._authenticate_oauth(auth_kwargs["client_id"], auth_kwargs["client_secret"]) self._authenticate_oauth(
auth_kwargs["client_id"], auth_kwargs["client_secret"]
)
else: else:
self._authenticate_user(auth_kwargs["username"], auth_kwargs["password"]) self._authenticate_user(auth_kwargs["username"], auth_kwargs["password"])