first ruggedy draft
This commit is contained in:
582
apps/local-agent/mousehold_agent/fints_connector.py
Normal file
582
apps/local-agent/mousehold_agent/fints_connector.py
Normal file
@@ -0,0 +1,582 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import getpass
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from html.parser import HTMLParser
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from fints.client import FinTS3PinTanClient, NeedTANResponse
|
||||
from fints.models import SEPAAccount
|
||||
from fints.utils import minimal_interactive_cli_bootstrap
|
||||
|
||||
DEFAULT_PROFILE = "default"
|
||||
DEFAULT_STATE_DIR = Path.home() / ".config" / "mousehold" / "fints"
|
||||
PRODUCT_ID_HELP = (
|
||||
"FinTS product_id is required by python-fints >=4. Register one with Deutsche Kreditwirtschaft/ZKA "
|
||||
"and provide it as MOUSEHOLD_FINTS_PRODUCT_ID, --product-id, or in the FinTS web form."
|
||||
)
|
||||
|
||||
|
||||
class FinTSConfigurationError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FinTSConfig:
|
||||
bank_identifier: str
|
||||
user_id: str
|
||||
pin: str
|
||||
server: str
|
||||
product_id: str
|
||||
customer_id: str | None = None
|
||||
product_version: str | None = None
|
||||
tan_mechanism: str | None = None
|
||||
tan_medium: str | None = None
|
||||
profile: str = DEFAULT_PROFILE
|
||||
state_dir: Path = DEFAULT_STATE_DIR
|
||||
store_state: bool = True
|
||||
interactive_bootstrap: bool = True
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FinTSFetchResult:
|
||||
account: SEPAAccount
|
||||
transactions: list[Any]
|
||||
observations: list[dict[str, Any]]
|
||||
|
||||
|
||||
class _HTMLTextExtractor(HTMLParser):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.parts: list[str] = []
|
||||
|
||||
def handle_data(self, data: str) -> None:
|
||||
stripped = data.strip()
|
||||
if stripped:
|
||||
self.parts.append(stripped)
|
||||
|
||||
def text(self) -> str:
|
||||
return " ".join(self.parts)
|
||||
|
||||
|
||||
class FinTSProfileStore:
|
||||
def __init__(self, state_dir: Path, profile: str) -> None:
|
||||
self.state_dir = state_dir
|
||||
self.profile = profile
|
||||
self.path = state_dir / f"{profile}.json"
|
||||
|
||||
def load_client_data(self) -> bytes | None:
|
||||
if not self.path.exists():
|
||||
return None
|
||||
data = json.loads(self.path.read_text(encoding="utf-8"))
|
||||
client_data = data.get("client_data")
|
||||
if not client_data:
|
||||
return None
|
||||
return base64.b64decode(client_data.encode("ascii"))
|
||||
|
||||
def save_client_data(self, client: FinTS3PinTanClient, config: FinTSConfig) -> None:
|
||||
self.state_dir.mkdir(parents=True, exist_ok=True)
|
||||
payload = {
|
||||
"profile": self.profile,
|
||||
"bank_identifier": config.bank_identifier,
|
||||
"server": config.server,
|
||||
"user_id": config.user_id,
|
||||
"customer_id": config.customer_id,
|
||||
"updated_at": datetime.now().isoformat(timespec="seconds"),
|
||||
"client_data": base64.b64encode(client.deconstruct()).decode("ascii"),
|
||||
}
|
||||
tmp_path = self.path.with_suffix(".tmp")
|
||||
tmp_path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
|
||||
os.chmod(tmp_path, 0o600)
|
||||
tmp_path.replace(self.path)
|
||||
os.chmod(self.path, 0o600)
|
||||
|
||||
|
||||
def config_from_args(args: Any) -> FinTSConfig:
|
||||
pin = getattr(args, "pin", None) or os.environ.get(getattr(args, "pin_env", "MOUSEHOLD_FINTS_PIN"))
|
||||
if not pin:
|
||||
pin = getpass.getpass("FinTS PIN: ")
|
||||
|
||||
product_id = _required_arg(args, "product_id", "MOUSEHOLD_FINTS_PRODUCT_ID", PRODUCT_ID_HELP)
|
||||
product_version = getattr(args, "product_version", None) or os.environ.get("MOUSEHOLD_FINTS_PRODUCT_VERSION")
|
||||
|
||||
return FinTSConfig(
|
||||
bank_identifier=_required_arg(args, "blz", "MOUSEHOLD_FINTS_BLZ"),
|
||||
user_id=_required_arg(args, "bank_user_id", "MOUSEHOLD_FINTS_USER_ID"),
|
||||
pin=pin,
|
||||
server=_required_arg(args, "server", "MOUSEHOLD_FINTS_SERVER"),
|
||||
customer_id=getattr(args, "customer_id", None) or os.environ.get("MOUSEHOLD_FINTS_CUSTOMER_ID"),
|
||||
product_id=product_id,
|
||||
product_version=product_version,
|
||||
tan_mechanism=getattr(args, "tan_mechanism", None) or os.environ.get("MOUSEHOLD_FINTS_TAN_MECHANISM"),
|
||||
tan_medium=getattr(args, "tan_medium", None) or os.environ.get("MOUSEHOLD_FINTS_TAN_MEDIUM"),
|
||||
profile=getattr(args, "profile", DEFAULT_PROFILE),
|
||||
state_dir=Path(getattr(args, "state_dir", DEFAULT_STATE_DIR)).expanduser(),
|
||||
store_state=not getattr(args, "no_state", False),
|
||||
interactive_bootstrap=not getattr(args, "no_bootstrap", False),
|
||||
)
|
||||
|
||||
|
||||
def create_client(config: FinTSConfig) -> FinTS3PinTanClient:
|
||||
if not config.product_id:
|
||||
raise FinTSConfigurationError(PRODUCT_ID_HELP)
|
||||
|
||||
kwargs: dict[str, Any] = {"product_id": config.product_id}
|
||||
if config.product_version:
|
||||
kwargs["product_version"] = config.product_version
|
||||
|
||||
store = FinTSProfileStore(config.state_dir, config.profile)
|
||||
if config.store_state and (client_data := store.load_client_data()):
|
||||
kwargs["from_data"] = client_data
|
||||
|
||||
return FinTS3PinTanClient(
|
||||
config.bank_identifier,
|
||||
config.user_id,
|
||||
config.pin,
|
||||
config.server,
|
||||
customer_id=config.customer_id,
|
||||
tan_medium=config.tan_medium,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
def bootstrap_client(client: FinTS3PinTanClient, config: FinTSConfig) -> None:
|
||||
if config.tan_mechanism and not client.get_current_tan_mechanism():
|
||||
client.set_tan_mechanism(config.tan_mechanism)
|
||||
if config.interactive_bootstrap:
|
||||
minimal_interactive_cli_bootstrap(client)
|
||||
|
||||
|
||||
def persist_client(client: FinTS3PinTanClient, config: FinTSConfig) -> None:
|
||||
if config.store_state:
|
||||
FinTSProfileStore(config.state_dir, config.profile).save_client_data(client, config)
|
||||
|
||||
|
||||
def with_fints_dialog(config: FinTSConfig, operation: Any) -> Any:
|
||||
client = create_client(config)
|
||||
bootstrap_client(client, config)
|
||||
try:
|
||||
with client:
|
||||
while isinstance(client.init_tan_response, NeedTANResponse):
|
||||
client.init_tan_response = answer_tan_challenge(client, client.init_tan_response, config)
|
||||
return operation(client)
|
||||
finally:
|
||||
persist_client(client, config)
|
||||
|
||||
|
||||
def list_accounts(config: FinTSConfig, *, include_balances: bool = False) -> list[dict[str, Any]]:
|
||||
def operation(client: FinTS3PinTanClient) -> list[dict[str, Any]]:
|
||||
rows: list[dict[str, Any]] = []
|
||||
for index, account in enumerate(client.get_sepa_accounts()):
|
||||
row = account_to_dict(account)
|
||||
row["index"] = index
|
||||
if include_balances:
|
||||
balance = resolve_tan_response(client, client.get_balance(account), config)
|
||||
row["balance"] = balance_to_dict(balance)
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
return with_fints_dialog(config, operation)
|
||||
|
||||
|
||||
def fetch_transactions(
|
||||
config: FinTSConfig,
|
||||
*,
|
||||
iban: str | None,
|
||||
start_date: date,
|
||||
end_date: date,
|
||||
include_pending: bool = False,
|
||||
household_id: str | None = None,
|
||||
source_account_id: str | None = None,
|
||||
owner_user_id: str | None = None,
|
||||
) -> FinTSFetchResult:
|
||||
def operation(client: FinTS3PinTanClient) -> FinTSFetchResult:
|
||||
account = select_account(client.get_sepa_accounts(), iban)
|
||||
response = client.get_transactions(
|
||||
account,
|
||||
start_date=start_date,
|
||||
end_date=end_date,
|
||||
include_pending=include_pending,
|
||||
)
|
||||
transactions = resolve_tan_response(client, response, config)
|
||||
observations = [
|
||||
transaction_to_observation(
|
||||
transaction,
|
||||
account=account,
|
||||
household_id=household_id,
|
||||
source_account_id=source_account_id,
|
||||
owner_user_id=owner_user_id,
|
||||
)
|
||||
for transaction in transactions
|
||||
]
|
||||
return FinTSFetchResult(account=account, transactions=list(transactions), observations=observations)
|
||||
|
||||
return with_fints_dialog(config, operation)
|
||||
|
||||
|
||||
def resolve_tan_response(
|
||||
client: FinTS3PinTanClient,
|
||||
response: Any,
|
||||
config: FinTSConfig,
|
||||
) -> Any:
|
||||
while isinstance(response, NeedTANResponse):
|
||||
response = answer_tan_challenge(client, response, config)
|
||||
return response
|
||||
|
||||
|
||||
def answer_tan_challenge(
|
||||
client: FinTS3PinTanClient,
|
||||
response: NeedTANResponse,
|
||||
config: FinTSConfig,
|
||||
) -> Any:
|
||||
print("\nTAN required by bank.")
|
||||
challenge_text = challenge_to_text(response)
|
||||
if challenge_text:
|
||||
print(challenge_text)
|
||||
|
||||
challenge_file = write_binary_challenge(response, config)
|
||||
if challenge_file:
|
||||
print(f"Challenge file: {challenge_file}")
|
||||
|
||||
if getattr(response, "challenge_hhduc", None):
|
||||
print("A chipTAN/flicker challenge is available.")
|
||||
print("Use your bank reader/app if it can read the challenge shown by your terminal.")
|
||||
|
||||
if getattr(response, "decoupled", False):
|
||||
input("Confirm the login/order in your banking app, then press Enter here.")
|
||||
return client.send_tan(response, "")
|
||||
|
||||
tan = getpass.getpass("TAN: ")
|
||||
return client.send_tan(response, tan)
|
||||
|
||||
|
||||
def tan_challenge_to_dict(response: NeedTANResponse) -> dict[str, Any]:
|
||||
matrix = getattr(response, "challenge_matrix", None)
|
||||
matrix_payload = None
|
||||
if matrix:
|
||||
mime_type, payload = matrix
|
||||
matrix_payload = {
|
||||
"mime_type": str(mime_type),
|
||||
"data_base64": base64.b64encode(payload).decode("ascii"),
|
||||
}
|
||||
return {
|
||||
"challenge": challenge_to_text(response),
|
||||
"challenge_html": str(getattr(response, "challenge_html", "") or ""),
|
||||
"challenge_hhduc": getattr(response, "challenge_hhduc", None),
|
||||
"challenge_matrix": matrix_payload,
|
||||
"decoupled": bool(getattr(response, "decoupled", False)),
|
||||
}
|
||||
|
||||
|
||||
def challenge_to_text(response: NeedTANResponse) -> str:
|
||||
if getattr(response, "challenge", None):
|
||||
return str(response.challenge)
|
||||
if getattr(response, "challenge_html", None):
|
||||
parser = _HTMLTextExtractor()
|
||||
parser.feed(str(response.challenge_html))
|
||||
return parser.text()
|
||||
if getattr(response, "challenge_raw", None):
|
||||
return str(response.challenge_raw)
|
||||
return ""
|
||||
|
||||
|
||||
def write_binary_challenge(response: NeedTANResponse, config: FinTSConfig) -> Path | None:
|
||||
matrix = getattr(response, "challenge_matrix", None)
|
||||
if not matrix:
|
||||
return None
|
||||
|
||||
mime_type, payload = matrix
|
||||
suffix = ".bin"
|
||||
if "png" in str(mime_type).lower():
|
||||
suffix = ".png"
|
||||
elif "jpeg" in str(mime_type).lower() or "jpg" in str(mime_type).lower():
|
||||
suffix = ".jpg"
|
||||
|
||||
output_dir = config.state_dir / "challenges"
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
path = output_dir / f"{config.profile}-{datetime.now().strftime('%Y%m%d%H%M%S')}{suffix}"
|
||||
path.write_bytes(payload)
|
||||
os.chmod(path, 0o600)
|
||||
return path
|
||||
|
||||
|
||||
def select_account(accounts: list[SEPAAccount], iban: str | None) -> SEPAAccount:
|
||||
if not accounts:
|
||||
raise RuntimeError("No SEPA accounts returned by the bank.")
|
||||
if not iban:
|
||||
if len(accounts) > 1:
|
||||
print("Multiple accounts returned; using the first one. Pass --iban to select explicitly.")
|
||||
return accounts[0]
|
||||
normalized = iban.replace(" ", "").upper()
|
||||
for account in accounts:
|
||||
if (account.iban or "").replace(" ", "").upper() == normalized:
|
||||
return account
|
||||
available = ", ".join(account.iban or "<no iban>" for account in accounts)
|
||||
raise RuntimeError(f"IBAN not found. Available accounts: {available}")
|
||||
|
||||
|
||||
def date_range_from_args(args: Any) -> tuple[date, date]:
|
||||
end_date = _parse_date_arg(getattr(args, "end_date", None)) or date.today()
|
||||
start_date = _parse_date_arg(getattr(args, "start_date", None))
|
||||
if not start_date:
|
||||
start_date = end_date - timedelta(days=getattr(args, "days", 30))
|
||||
return start_date, end_date
|
||||
|
||||
|
||||
def transaction_to_observation(
|
||||
transaction: Any,
|
||||
*,
|
||||
account: SEPAAccount,
|
||||
household_id: str | None,
|
||||
source_account_id: str | None,
|
||||
owner_user_id: str | None,
|
||||
) -> dict[str, Any]:
|
||||
raw_data = transaction_data(transaction)
|
||||
normalized_data = jsonable(raw_data)
|
||||
amount, currency = extract_amount(raw_data)
|
||||
booking_date = extract_date(raw_data, "date", "booking_date", "entry_date")
|
||||
value_date = extract_date(raw_data, "entry_date", "valuta_date", "date")
|
||||
counterparty = pick_text(
|
||||
raw_data,
|
||||
"applicant_name",
|
||||
"applicant_iban",
|
||||
"applicant_bin",
|
||||
"counterparty",
|
||||
"name",
|
||||
)
|
||||
description = pick_text(
|
||||
raw_data,
|
||||
"purpose",
|
||||
"transaction_details",
|
||||
"extra_details",
|
||||
"posting_text",
|
||||
"prima_nota",
|
||||
"additional_position_reference",
|
||||
)
|
||||
reference = pick_text(
|
||||
raw_data,
|
||||
"id",
|
||||
"transaction_reference",
|
||||
"customer_reference",
|
||||
"bank_reference",
|
||||
"end_to_end_reference",
|
||||
"eref",
|
||||
"mref",
|
||||
"svwz",
|
||||
)
|
||||
fallback_reference = stable_transaction_reference(account, normalized_data)
|
||||
signed_amount = amount
|
||||
|
||||
return {
|
||||
"household_id": household_id,
|
||||
"source_type": "fints",
|
||||
"source_account_id": source_account_id,
|
||||
"owner_user_id": owner_user_id,
|
||||
"source_reference": reference or fallback_reference,
|
||||
"provider_transaction_id": reference or fallback_reference,
|
||||
"raw_payload_json": {
|
||||
"account": account_to_dict(account),
|
||||
"transaction": normalized_data,
|
||||
"source": "fints",
|
||||
},
|
||||
"amount": str(abs(signed_amount)),
|
||||
"currency": currency or "EUR",
|
||||
"event_date": (booking_date or value_date or date.today()).isoformat(),
|
||||
"booking_date": booking_date.isoformat() if booking_date else None,
|
||||
"merchant": counterparty,
|
||||
"counterparty": counterparty,
|
||||
"description": description,
|
||||
"event_type": infer_event_type(signed_amount, counterparty, description),
|
||||
"status": "confirmed",
|
||||
}
|
||||
|
||||
|
||||
def transaction_data(transaction: Any) -> dict[str, Any]:
|
||||
data = getattr(transaction, "data", transaction)
|
||||
if isinstance(data, dict):
|
||||
return data
|
||||
if hasattr(data, "_asdict"):
|
||||
return data._asdict()
|
||||
return {"value": data}
|
||||
|
||||
|
||||
def extract_amount(data: dict[str, Any]) -> tuple[Decimal, str | None]:
|
||||
value = data.get("amount") or data.get("transaction_amount")
|
||||
currency = data.get("currency")
|
||||
if hasattr(value, "amount"):
|
||||
currency = getattr(value, "currency", currency)
|
||||
return Decimal(str(value.amount)), currency
|
||||
if isinstance(value, dict):
|
||||
amount = value.get("amount") or value.get("value")
|
||||
currency = value.get("currency", currency)
|
||||
return Decimal(str(amount)), currency
|
||||
if value is None:
|
||||
raise RuntimeError(f"Transaction has no amount: {data}")
|
||||
return Decimal(str(value).replace(",", ".")), currency
|
||||
|
||||
|
||||
def extract_date(data: dict[str, Any], *keys: str) -> date | None:
|
||||
for key in keys:
|
||||
value = data.get(key)
|
||||
if isinstance(value, date):
|
||||
return value
|
||||
if isinstance(value, datetime):
|
||||
return value.date()
|
||||
if isinstance(value, str) and value:
|
||||
for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%Y%m%d"):
|
||||
try:
|
||||
return datetime.strptime(value, fmt).date()
|
||||
except ValueError:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def pick_text(data: dict[str, Any], *keys: str) -> str | None:
|
||||
values: list[str] = []
|
||||
for key in keys:
|
||||
value = data.get(key)
|
||||
if value is None or value == "":
|
||||
continue
|
||||
if isinstance(value, list | tuple):
|
||||
values.extend(str(item).strip() for item in value if str(item).strip())
|
||||
else:
|
||||
values.append(str(value).strip())
|
||||
if not values:
|
||||
return None
|
||||
return " ".join(dict.fromkeys(values))[:500]
|
||||
|
||||
|
||||
def infer_event_type(amount: Decimal, counterparty: str | None, description: str | None) -> str:
|
||||
if amount >= 0:
|
||||
return "income"
|
||||
combined = f"{counterparty or ''} {description or ''}".lower()
|
||||
liability_markers = (
|
||||
"paypal",
|
||||
"klarna",
|
||||
"kreditkarte",
|
||||
"credit card",
|
||||
"visa",
|
||||
"mastercard",
|
||||
"american express",
|
||||
"amex",
|
||||
)
|
||||
if any(marker in combined for marker in liability_markers):
|
||||
return "liability_payment"
|
||||
return "expense"
|
||||
|
||||
|
||||
def account_to_dict(account: SEPAAccount) -> dict[str, Any]:
|
||||
return {
|
||||
"iban": account.iban,
|
||||
"bic": account.bic,
|
||||
"accountnumber": account.accountnumber,
|
||||
"subaccount": account.subaccount,
|
||||
"blz": account.blz,
|
||||
}
|
||||
|
||||
|
||||
def balance_to_dict(balance: Any) -> dict[str, Any]:
|
||||
amount = getattr(balance, "amount", None)
|
||||
return {
|
||||
"amount": str(getattr(amount, "amount", amount)),
|
||||
"currency": getattr(amount, "currency", None),
|
||||
"date": getattr(balance, "date", None).isoformat()
|
||||
if isinstance(getattr(balance, "date", None), date)
|
||||
else str(getattr(balance, "date", "")),
|
||||
"status": getattr(balance, "status", None),
|
||||
}
|
||||
|
||||
|
||||
def stable_transaction_reference(account: SEPAAccount, payload: dict[str, Any]) -> str:
|
||||
digest = hashlib.sha256(
|
||||
json.dumps(
|
||||
{"account": account_to_dict(account), "transaction": payload},
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode("utf-8")
|
||||
).hexdigest()
|
||||
return f"fints:{account.iban or account.accountnumber}:{digest}"
|
||||
|
||||
|
||||
def jsonable(value: Any) -> Any:
|
||||
if isinstance(value, dict):
|
||||
return {str(key): jsonable(item) for key, item in value.items()}
|
||||
if isinstance(value, list | tuple | set):
|
||||
return [jsonable(item) for item in value]
|
||||
if isinstance(value, datetime | date):
|
||||
return value.isoformat()
|
||||
if isinstance(value, Decimal):
|
||||
return str(value)
|
||||
if hasattr(value, "amount"):
|
||||
return {
|
||||
"amount": str(value.amount),
|
||||
"currency": getattr(value, "currency", None),
|
||||
}
|
||||
if hasattr(value, "_asdict"):
|
||||
return jsonable(value._asdict())
|
||||
return value if isinstance(value, str | int | float | bool | type(None)) else str(value)
|
||||
|
||||
|
||||
def display_accounts(accounts: list[dict[str, Any]]) -> None:
|
||||
rows = []
|
||||
for account in accounts:
|
||||
balance = account.get("balance")
|
||||
balance_text = ""
|
||||
if isinstance(balance, dict) and balance.get("amount"):
|
||||
balance_text = f"{balance['amount']} {balance.get('currency') or ''}".strip()
|
||||
rows.append(
|
||||
[
|
||||
str(account.get("index", "")),
|
||||
account.get("iban") or "",
|
||||
account.get("bic") or "",
|
||||
account.get("accountnumber") or "",
|
||||
balance_text,
|
||||
]
|
||||
)
|
||||
print_table(["#", "IBAN", "BIC", "Account", "Balance"], rows)
|
||||
|
||||
|
||||
def display_transactions(observations: list[dict[str, Any]]) -> None:
|
||||
rows = [
|
||||
[
|
||||
observation.get("event_date") or "",
|
||||
observation.get("event_type") or "",
|
||||
observation.get("amount") or "",
|
||||
observation.get("currency") or "",
|
||||
observation.get("counterparty") or "",
|
||||
observation.get("description") or "",
|
||||
]
|
||||
for observation in observations
|
||||
]
|
||||
print_table(["Date", "Type", "Amount", "Cur", "Counterparty", "Description"], rows)
|
||||
|
||||
|
||||
def print_table(headers: list[str], rows: list[list[str]]) -> None:
|
||||
widths = [
|
||||
max(len(headers[index]), *(len(row[index]) for row in rows)) if rows else len(headers[index])
|
||||
for index in range(len(headers))
|
||||
]
|
||||
print(" ".join(header.ljust(widths[index]) for index, header in enumerate(headers)))
|
||||
print(" ".join("-" * width for width in widths))
|
||||
for row in rows:
|
||||
print(" ".join(row[index].ljust(widths[index]) for index in range(len(headers))))
|
||||
|
||||
|
||||
def _parse_date_arg(value: str | None) -> date | None:
|
||||
if not value:
|
||||
return None
|
||||
return date.fromisoformat(value)
|
||||
|
||||
|
||||
def _required_arg(args: Any, attribute: str, env_name: str, message: str | None = None) -> str:
|
||||
value = getattr(args, attribute, None) or os.environ.get(env_name)
|
||||
if not value:
|
||||
raise RuntimeError(message or f"{attribute.replace('_', '-')} or {env_name} is required")
|
||||
return value
|
||||
Reference in New Issue
Block a user