1
0
Fork 0
mirror of https://codeberg.org/Reuh/feather.git synced 2025-10-27 10:09:32 +00:00

refactor: add missing type hints

This commit is contained in:
Étienne Fildadut 2025-10-11 17:38:48 +02:00
parent 0fd5ec6458
commit 70b930a820
4 changed files with 32 additions and 43 deletions

View file

@ -12,7 +12,9 @@ from feather.data import Article, ArticleId, Category
class ClientSession(ABC): class ClientSession(ABC):
config: Config @abstractmethod
def __init__(self, config: Config):
self.config: Config = config
@abstractmethod @abstractmethod
def set_read_flag(self, article_ids: list[ArticleId], read: bool = True): def set_read_flag(self, article_ids: list[ArticleId], read: bool = True):
@ -36,21 +38,19 @@ class ClientSession(ABC):
pass pass
label_name = re.compile("user/.*/label/(.*)") label_name_re = re.compile("user/.*/label/(.*)")
class GReaderSession(ClientSession): class GReaderSession(ClientSession):
"""Google Reader API client""" """Google Reader API client"""
greader: google_reader.Client
auth_token: str
csrf_token: str
def __init__(self, config: Config): def __init__(self, config: Config):
self.config = config self.config: Config = config
self.greader = google_reader.Client(config.server_url) self.greader: google_reader.Client = google_reader.Client(config.server_url)
self.auth_token = self.greader.login(config.server_user, config.server_password) self.auth_token: str = self.greader.login(
self.csrf_token = self.greader.get_token(self.auth_token) config.server_user, config.server_password
)
self.csrf_token: str = self.greader.get_token(self.auth_token)
def set_read_flag(self, article_ids: list[ArticleId], read: bool = True): def set_read_flag(self, article_ids: list[ArticleId], read: bool = True):
if read: if read:
@ -76,7 +76,7 @@ class GReaderSession(ClientSession):
] ]
categories = [] categories = []
for category in tags: for category in tags:
category_name = category.label or label_name.search(category.id).group(1) category_name = category.label or label_name_re.search(category.id).group(1)
category_id = category.id category_id = category.id
categories.append(Category(id=category_id, title=category_name)) categories.append(Category(id=category_id, title=category_name))
return categories return categories
@ -134,19 +134,16 @@ class GReaderArticle(Article):
class TTRSession(ClientSession): class TTRSession(ClientSession):
"""Tiny Tiny RSS API client""" """Tiny Tiny RSS API client"""
ttrss: TTRClient
feeds: dict
def __init__(self, config: Config): def __init__(self, config: Config):
self.config = config self.config: Config = config
self.ttrss = TTRClient( self.ttrss: TTRClient = TTRClient(
config.server_url, config.server_url,
config.server_user, config.server_user,
config.server_password, config.server_password,
auto_login=True, auto_login=True,
) )
self.ttrss.login() self.ttrss.login()
self.feeds = {} self.feeds: dict = {}
def set_read_flag(self, article_ids: list[ArticleId], read: bool = True): def set_read_flag(self, article_ids: list[ArticleId], read: bool = True):
if read: if read:

View file

@ -6,14 +6,13 @@ from zoneinfo import ZoneInfo
from pathlib import Path from pathlib import Path
from jinja2 import Template from jinja2 import Template
default_config_path: Path = Path(__file__).parent / "config.default.toml"
class ConfigurationError(ValueError): class ConfigurationError(ValueError):
pass pass
default_config_path = Path(__file__).parent / "config.default.toml"
class Config: class Config:
def __init__(self): def __init__(self):
with default_config_path.open("rb") as f: with default_config_path.open("rb") as f:

View file

@ -48,20 +48,17 @@ type CategoryId = int | str
class Category: class Category:
id: CategoryId # category id def fromdict(d) -> Category:
title: str # category name
parents: list[Category] # list of parent categories
order: int = 0 # category display order, starting from 1 (0 if unknown)
def fromdict(d):
parents = [Category.fromdict(parent) for parent in d["parents"]] parents = [Category.fromdict(parent) for parent in d["parents"]]
return Category(d["id"], d["title"], parents, d["order"]) return Category(d["id"], d["title"], parents, d["order"])
def __init__(self, id, title, parents=[], order=0): def __init__(self, id, title, parents=[], order=0):
self.id = id self.id: CategoryId = id # category unique id
self.title = title self.title: str = title # category name
self.parents = parents self.parents: list[Category] = parents # list of parent categories
self.order = order self.order: int = (
order # category display order, starting from 1 (0 if unknown)
)
def asdict(self): def asdict(self):
return { return {
@ -106,7 +103,7 @@ class Article(ABC):
language: str = "" # article language language: str = "" # article language
image_url: str = "" # article main image image_url: str = "" # article main image
def _get_html_path(self): def _get_html_path(self) -> Path:
config = self.config config = self.config
# Category directory path # Category directory path
@ -205,7 +202,7 @@ class Article(ABC):
"""Delete the JSON file associated with this article.""" """Delete the JSON file associated with this article."""
self.json_path.unlink() self.json_path.unlink()
def has_html(self): def has_html(self) -> bool:
"""Check if the HTML file associated with the article exists on disk.""" """Check if the HTML file associated with the article exists on disk."""
if self.html_path is None: if self.html_path is None:
return False return False
@ -257,7 +254,7 @@ class Article(ABC):
self.compute_fields() # recompute formatted datetime & paths from the current configuration self.compute_fields() # recompute formatted datetime & paths from the current configuration
self.write() # rewrite JSON & HTML self.write() # rewrite JSON & HTML
def was_updated(self, old_article: Article): def was_updated(self, old_article: Article) -> bool:
"""Returns true if the article is different from a previous version in a way that would require regeneration""" """Returns true if the article is different from a previous version in a way that would require regeneration"""
return old_article._get_template_dict() != self._get_template_dict() return old_article._get_template_dict() != self._get_template_dict()

View file

@ -2,20 +2,17 @@
import asyncio import asyncio
import signal import signal
from typing import Iterable
from feather.config import Config from feather.config import Config
from feather.client import GReaderSession, TTRSession, ClientSession, ArticleId from feather.client import GReaderSession, TTRSession, ClientSession, Article, ArticleId
from feather.data import FileArticle from feather.data import FileArticle
class FeatherApp: class FeatherApp:
config: Config
def __init__(self, config: Config): def __init__(self, config: Config):
self.config = config self.config: Config = config
self._client_session = None self._client_session: ClientSession = None
_client_session: ClientSession
def get_client_session(self) -> ClientSession: def get_client_session(self) -> ClientSession:
"""Connect to the server and return a ClientSession object; return an existing ClientSession if we are already connected""" """Connect to the server and return a ClientSession object; return an existing ClientSession if we are already connected"""
@ -32,7 +29,7 @@ class FeatherApp:
) )
return self._client_session return self._client_session
def iter_articles(self): def iter_articles(self) -> Iterable[Article]:
"""Iterate over all the articles in local storage""" """Iterate over all the articles in local storage"""
config = self.config config = self.config
for json_path in config.json_root.glob("*.json"): for json_path in config.json_root.glob("*.json"):
@ -175,9 +172,8 @@ class FeatherApp:
async def daemon(self): async def daemon(self):
"""Start the synchronization daemon""" """Start the synchronization daemon"""
config = self.config
print( print(
f"Started in daemon mode; changes will be downloaded from the server every {config.daemon_sync_down_every}s and uploaded every {config.daemon_sync_up_every}s" f"Started in daemon mode; changes will be downloaded from the server every {self.config.daemon_sync_down_every}s and uploaded every {self.config.daemon_sync_up_every}s"
) )
async with asyncio.TaskGroup() as tg: async with asyncio.TaskGroup() as tg:
tup = tg.create_task(self.daemon_sync_up_loop()) tup = tg.create_task(self.daemon_sync_up_loop())