49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
# EnergyPriceScraperFactory.py
|
|
from __future__ import annotations
|
|
import importlib
|
|
from typing import Any, cast, Type
|
|
from EnergyPriceScraper import EnergyPriceScraperBase
|
|
|
|
import os
|
|
import psycopg
|
|
|
|
DB_HOST = os.getenv("PGHOST", "192.168.30.10")
|
|
DB_PORT = int(os.getenv("PGPORT", "5432"))
|
|
DB_NAME = os.getenv("PGDATABASE", "postgres")
|
|
DB_USER = os.getenv("PGUSER", "energy_ingest")
|
|
DB_PASS = os.getenv("PGPASSWORD", "2f1rLCa03mQrbmlCbD6envk")
|
|
|
|
def setup_db():
|
|
# psycopg 3
|
|
conn = psycopg.connect(
|
|
host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASS
|
|
)
|
|
return conn
|
|
|
|
def create(name: str, /, **kwargs: Any) -> EnergyPriceScraperBase:
|
|
"""
|
|
Convention:
|
|
module: Scraper.<Name>Scraper
|
|
class: <Name>Provider
|
|
Example: create("TauronG13", rates={...})
|
|
"""
|
|
safe = "".join(ch for ch in name if ch.isalnum() or ch == "_")
|
|
module_name = f"Scraper.{safe}Scraper"
|
|
class_name = f"{safe}Scraper"
|
|
|
|
try:
|
|
mod = importlib.import_module(module_name)
|
|
except ModuleNotFoundError as e:
|
|
raise ValueError(f"Scraper module not found: {module_name}") from e
|
|
|
|
try:
|
|
cls = getattr(mod, class_name)
|
|
except AttributeError as e:
|
|
raise ValueError(f"Scraper class not found: {class_name} in {module_name}") from e
|
|
|
|
if not issubclass(cls, EnergyPriceScraperBase):
|
|
raise TypeError(f"{class_name} must inherit PriceScraperBase")
|
|
|
|
ProviderCls = cast(Type[EnergyPriceScraperBase], cls)
|
|
return ProviderCls(**kwargs) # type: ignore[arg-type]
|