from config import Config
from .base import BaseUtil
from typing import TypedDict, Literal, Dict, Any, Tuple, List, Callable
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime
import time
from functools import wraps
from util.format import FormatUtil
from airflow.models.connection import Connection
### Types
class SBOFabricVersion(TypedDict):
state_fips: str
state_abbr: str
fabric_version: str
"""Fabric addresses version. it is date related to 2, 3, 4
"""
fabric_ready_version: int
"""Ready fabric version, 19, 20,21 etc
"""
tile_last_update: str
map_type: Literal["cpc"]
class FabricVersion(TypedDict):
version_date: str
"""Fabric addresses version. it is date related to v2, v3, v4
"""
update_date: str
""" date and time fabric fixed availability ingested
"""
version: int
"""Ready fabric version, 19, 20,21 etc
"""
process_uuid: str
[docs]
class FCCUtil:
FCC_API_KEYS = "fcc_api_creds"
# FCC Funding Map per-state file themes — files named state{fips}_*.csv
# split into these themes by file-name suffix (DATASME-1997); the two
# funded-locations categories first appeared in the 2026-06-25 release
FCC_FM_UNSERVED_UNFUNDED = "unserved_unfunded"
FCC_FM_FUNDED_LOCATIONS = "funded_locations"
FCC_FM_PROGRAM_FUNDED_LOCATIONS = "program_funded_locations"
FCC_FM_STATE_FILE_THEMES = [
FCC_FM_UNSERVED_UNFUNDED,
FCC_FM_FUNDED_LOCATIONS,
FCC_FM_PROGRAM_FUNDED_LOCATIONS,
]
[docs]
@staticmethod
def fcc_fm_state_file_theme(file_name: str) -> str:
"""
Map an FCC Funding Map state{fips}_* file name to its theme (DATASME-1997):
- state{fips}_unserved_unfunded_* -> unserved_unfunded
- state{fips}_funded_locations_* -> funded_locations
- state{fips}_program{id}_funded_locations_* -> program_funded_locations
Raises AirflowException on anything else so a new FCC file category
fails the run loudly instead of landing in the wrong theme.
:param file_name: the file name (with or without extension)
:return: the theme directory name
"""
from airflow.exceptions import AirflowException
suffix = file_name.split("_", 1)[1] if "_" in file_name else ""
if suffix.startswith(FCCUtil.FCC_FM_UNSERVED_UNFUNDED):
return FCCUtil.FCC_FM_UNSERVED_UNFUNDED
if suffix.startswith(FCCUtil.FCC_FM_FUNDED_LOCATIONS):
return FCCUtil.FCC_FM_FUNDED_LOCATIONS
if suffix.startswith("program") and FCCUtil.FCC_FM_FUNDED_LOCATIONS in suffix:
return FCCUtil.FCC_FM_PROGRAM_FUNDED_LOCATIONS
raise AirflowException(
f"Unrecognized state file name '{file_name}' — cannot classify into a theme. "
f"If the FCC added a new per-state file category, add it to "
f"FCCUtil.FCC_FM_STATE_FILE_THEMES and fcc_fm_state_file_theme(), "
f"with matching cet_parquet_r_fcc_fm_/t_fcc_fm_ SQL."
)
[docs]
@staticmethod
def create_fcc_api_caller(
max_per_minute: int = 10,
max_attempts: int = 3,
):
import requests
from requests.exceptions import ChunkedEncodingError, RequestException
from tenacity import (
retry,
wait_exponential,
stop_after_attempt,
retry_if_exception_type,
RetryError,
)
min_interval = 60.0 / max_per_minute
last_called = [0.0]
def rate_limited(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
left_to_wait = min_interval - elapsed
if left_to_wait > 0:
time.sleep(left_to_wait)
ret = func(*args, **kwargs)
last_called[0] = time.time()
return ret
return wrapper
@rate_limited
@retry(
wait=wait_exponential(multiplier=1, min=4, max=10),
stop=stop_after_attempt(max_attempts),
retry=retry_if_exception_type((ChunkedEncodingError, RequestException)),
reraise=True,
)
def api_call(
url: str, headers: Dict[str, str], return_json: bool = False
) -> Any:
response = requests.get(url, headers=headers, stream=True)
response.raise_for_status()
return response.json() if return_json else response.content
def api_call_with_stats(*args, **kwargs) -> Tuple[Any, List[str]]:
try:
content = api_call(*args, **kwargs)
return content, []
except RetryError as e:
error_messages = [
f"Attempt {i + 1}: {str(attempt.exception())}"
for i, attempt in enumerate(e.retry_state.attempts)
if attempt.failed
]
return None, error_messages
except Exception as e:
return None, [str(e)]
return api_call_with_stats
[docs]
@staticmethod
def create_fcc_api_client(
max_per_minute: int = 10,
max_attempts: int = 3,
) -> Callable[[str, bool], Tuple[Any, List[str]]]:
api_call = FCCUtil.create_fcc_api_caller(max_per_minute, max_attempts)
headers = FCCUtil.gen_fcc_headers()
def fcc_api_client(
url: str, return_json: bool = False
) -> Tuple[Any, List[str]]:
return api_call(url, headers, return_json)
return fcc_api_client
[docs]
@staticmethod
def get_max_filing_subtype(data: list[dict]) -> dict[str, str]:
"""
Get the max filing subtype from fcc broadband map
:param data: the data to get the max filing subtype from
:return: the max filing subtype
"""
return max(
data, key=lambda x: datetime.strptime(x["filing_subtype"], "%B %d, %Y")
)
[docs]
@staticmethod
def get_current_process_uuid(
external_request: bool = False,
proxy_request: bool = False,
) -> str:
"""
Get the current process uuid from fcc broadband map
:param external_request: whether or not to use external request
:param proxy_request: whether or not to use proxy request
:return: the current process uuid
"""
request = (
"external_get_request"
if external_request
else "proxy_get_request" if proxy_request else "get_request"
)
resp = getattr(BaseUtil, request)(Config.FCC_PROCESS_UUID_URL)
current_data = FCCUtil.get_max_filing_subtype(resp.json()["data"])
return current_data["process_uuid"]
[docs]
@staticmethod
def get_fcc_last_updated(
process_uuid: str,
external_request: bool = False,
proxy_request: bool = False,
as_datetime: bool = True,
override_key: str = "last_updated_date", # or data_as_of_date
):
"""
Get the last updated date from fcc broadband map
:param process_uuid: the process uuid
:param external_request: whether or not to use external request
:param proxy_request: whether or not to use proxy request
:param as_datetime: whether or not to return as datetime or string
:param override_key: the override key
:return: the last updated date, or data as of date
"""
request = (
"external_get_request"
if external_request
else "proxy_get_request" if proxy_request else "get_request"
)
resp = getattr(BaseUtil, request)(
Config.FCC_UPDATED_URL.format(process_uuid=process_uuid)
)
date_str = resp.json()["data"][0][override_key]
if as_datetime:
try:
return datetime.strptime(date_str, Config.FCC_DATE_FMT)
except ValueError:
return datetime.strptime(date_str, Config.FCC_DATE_FMT2)
else:
return date_str
[docs]
@staticmethod
def ping_fcc_website(
external_request: bool = False,
proxy_request: bool = False,
max_retries: int = 10, # 10 retries
retry_interval_minutes: int = 1, # 1 minute
) -> bool:
"""
Check if the FCC website is available before making requests with retry capability.
Need this because the FCC website is often appearing down but is up after a few minutes.
:param external_request: whether or not to use external request
:param proxy_request: whether or not to use proxy request
:param timeout: timeout in seconds for the request
:param max_retries: maximum number of retry attempts
:param retry_interval_minutes: time to wait between retries in minutes
:return: True if website is available, False otherwise
"""
import time
for attempt in range(1, max_retries + 1):
try:
request = (
"external_get_request"
if external_request
else "proxy_get_request" if proxy_request else "get_request"
)
resp = getattr(BaseUtil, request)(
Config.FCC_FUNDING_PUBLISHED_RUN_URL,
)
if resp.status_code == 200:
return True
print(
f"Attempt {attempt}/{max_retries}: FCC website returned status code {resp.status_code}"
)
except Exception as e:
print(
f"Attempt {attempt}/{max_retries}: Error connecting to FCC website: {str(e)}"
)
if attempt < max_retries:
sleep_seconds = retry_interval_minutes * 60
print(f"Waiting {retry_interval_minutes} minutes before next retry...")
time.sleep(sleep_seconds)
return False
[docs]
@staticmethod
def get_fcc_funding_info(
external_request: bool = False,
proxy_request: bool = False,
) -> tuple[str, str]:
"""
Get both the current funding process uuid and last updated date from fcc funding map
:param external_request: whether or not to use external request
:param proxy_request: whether or not to use proxy request
:return: tuple of (process_uuid, last_updated_date)
:raises ConnectionError: when website is not available
"""
if not FCCUtil.ping_fcc_website(external_request, proxy_request):
raise ConnectionError("FCC website is currently unavailable")
request = (
"external_get_request"
if external_request
else "proxy_get_request" if proxy_request else "get_request"
)
resp = getattr(BaseUtil, request)(Config.FCC_FUNDING_PUBLISHED_RUN_URL)
# Extract both pieces of information from the same response
data = resp.json()["data"][0]
process_uuid = data["process_uuid"]
last_updated = data["create_date"]
return process_uuid, last_updated
[docs]
@staticmethod
def get_latest_fabric_version(
database_name: str = Config.GAME_PG,
fa_ver: int = None,
fs_ver: int = None,
schema: str = "data",
fabric_version_tbname: str = "fabric_versions",
) -> FabricVersion:
"""gets latest published fabric version from PG TABLE: fabric_versions
:param database_name: _description_, defaults to Config.GAME_PG
:type database_name: str, optional
:param schema: _description_, defaults to "data"
:type schema: str, optional
:param fa_ver: the fabric address version, defaults to None
:type fa_ver: int, optional
:param fs_ver: the fabric fixed availability version, defaults to None; will take precendence over fa_ver if both are provided
:type fs_ver: int, optional
:return: an object with following structure version_date, update_date, version, process_uuid
:rtype: FabricVersion
"""
db_hook = PostgresHook(postgres_conn_id=database_name)
VERSION_DATE_DICT = {
1: "2023-06-30",
2: "2022-12-31",
3: "2023-06-30",
4: "2023-12-31",
5: "2024-06-30",
}
version_date = VERSION_DATE_DICT.get(int(fa_ver) if fa_ver else "", "")
where_cond = f" WHERE version_date='{version_date}' " if version_date else ""
# if fs_ver is provided, it will take precedence over fa_ver
where_cond = f" WHERE version = {int(fs_ver)} " if fs_ver else where_cond
sql_query = f"""
SELECT version_date, update_date, version, process_uuid
FROM {schema}.{fabric_version_tbname}
{where_cond}
ORDER BY version DESC LIMIT 1
"""
result = db_hook.get_first(sql_query)
if result:
return {
"version_date": FormatUtil.to_date_string(result[0]),
"update_date": FormatUtil.to_date_string(result[1]),
"version": int(result[2]),
"process_uuid": str(result[3]),
}