utils.py

import os
from datetime import date, timezone
from time import sleep
from typing import TypeAlias, Callable

from dotenv import load_dotenv
from dateutil import parser as dateparser
from atproto_client.request import Response

APICall: TypeAlias = Callable[[str, str | None, int | None], Response]


def dumb_retry(call: APICall, max_attempts: int = 25, wait: float = 2.0, inc_factor: float = 0.4) -> APICall:
    def wrapper(*args, **kwargs):
        w = wait
        for i in range(max_attempts):
            try:
                return call(*args, **kwargs)
            except Exception as e:
                print(f"Error: {e}\nRetrying in {w:.2f} seconds...")
                sleep(w)
                w += w * inc_factor
        else:
            raise RuntimeError(f"Failed after {max_attempts} attempts")

    return wrapper


def load_credentials(
    dotenv_path: str = ".env",
    username_var: str = "BLUESKY_USERNAME",
    password_var: str = "BLUESKY_PASSWORD",
) -> tuple[str, str]:
    load_dotenv(dotenv_path)
    username = os.getenv(username_var)
    password = os.getenv(password_var)
    if username is None or password is None:
        raise ValueError(f"Environment variables {username_var} and {password_var} must be set")
    return username, password


def parse_utc_date(timestamp: str) -> date:
    return dateparser.parse(timestamp).replace(tzinfo=timezone.utc).date()