feat(account): add database account fallback in MastodonApi.refresh_account

- Implemented _get_db_account to retrieve cached account data on fetch failure
- Enhanced error handling when refreshing account information
This commit is contained in:
Fabio Manganiello 2026-03-04 12:36:44 +01:00
commit 25d1d2b882
Signed by: blacklight
GPG key ID: D90FBA7F76362774

View file

@ -10,6 +10,7 @@ import requests
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from ..config import Config from ..config import Config
from ..db import Db
from ..errors import AccountNotFoundError, HttpError from ..errors import AccountNotFoundError, HttpError
from ..model import Account, Media, Post from ..model import Account, Media, Post
@ -22,6 +23,7 @@ class MastodonApi(ABC):
""" """
config: Config config: Config
db: Db
def _http_get(self, url: str, *args, **kwargs) -> dict: def _http_get(self, url: str, *args, **kwargs) -> dict:
""" """
@ -125,10 +127,20 @@ class MastodonApi(ABC):
@abstractmethod @abstractmethod
def get_campaign_url(self, account: Account) -> str | None: ... def get_campaign_url(self, account: Account) -> str | None: ...
def _get_db_account(self, account: Account) -> Account | None:
"""
Try to get the account from the database if available.
Returns None if db is not available or account not found.
"""
return self.db.get_account(account.url)
def refresh_account(self, account: Account) -> Account: def refresh_account(self, account: Account) -> Account:
""" """
Get the Mastodon ID of an account. Get the Mastodon ID of an account.
""" """
fetch_failed = False
account_info = {}
try: try:
if not account.id: if not account.id:
account.id = self._get_account_id(account) account.id = self._get_account_id(account)
@ -145,27 +157,40 @@ class MastodonApi(ABC):
str(exc), str(exc),
exc_info=True, exc_info=True,
) )
return account fetch_failed = True
try: if not fetch_failed:
account_info = self._http_get(account.api_url) try:
except Exception as exc: account_info = self._http_get(account.api_url)
log.warning( except Exception as exc:
"Failed to refresh account %s: %s", log.warning(
account.url, "Failed to refresh account %s: %s",
str(exc), account.url,
exc_info=True, str(exc),
) exc_info=True,
return account )
fetch_failed = True
if fetch_failed:
db_account = self._get_db_account(account)
if db_account:
log.info(
"Using cached account data for %s due to fetch failure",
account.url,
)
account = db_account
else:
return account
else:
account.id = str(account_info["id"])
account.display_name = account_info.get("display_name") or account.username
account.avatar_url = account_info.get("avatar_static")
account.header_url = account_info.get("header_static")
account.profile_note = account_info.get("note")
account.profile_fields = self._parse_profile_fields(account_info.get("fields", []))
account.disabled = bool(account_info["locked"])
account.created_at = self._convert_datetime(account_info["created_at"])
account.id = str(account_info["id"])
account.display_name = account_info.get("display_name") or account.username
account.avatar_url = account_info.get("avatar_static")
account.header_url = account_info.get("header_static")
account.profile_note = account_info.get("note")
account.profile_fields = self._parse_profile_fields(account_info.get("fields", []))
account.disabled = bool(account_info["locked"])
account.created_at = self._convert_datetime(account_info["created_at"])
campaign_url = self.get_campaign_url(account) campaign_url = self.get_campaign_url(account)
if campaign_url: if campaign_url:
account.campaign_url = campaign_url account.campaign_url = campaign_url