Add ruff

.pre-commit-config.yaml

@@ -1,4 +1,10 @@
 repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.1.11
+    hooks:
+      - id: ruff
+        args:
+          - --fix
   - repo: https://github.com/psf/black
     rev: 23.12.1
     hooks:

aio_georss_client/consts.py

@@ -1,63 +1,64 @@
 """Constants for feeds and feed entries."""
+from typing import Final
 
-ATTR_ATTRIBUTION = "attribution"
-CUSTOM_ATTRIBUTE = "custom_attribute"
+ATTR_ATTRIBUTION: Final = "attribution"
+CUSTOM_ATTRIBUTE: Final = "custom_attribute"
 
-DEFAULT_REQUEST_TIMEOUT = 10
+DEFAULT_REQUEST_TIMEOUT: Final = 10
 
-UPDATE_OK = "OK"
-UPDATE_OK_NO_DATA = "OK_NO_DATA"
-UPDATE_ERROR = "ERROR"
+UPDATE_OK: Final = "OK"
+UPDATE_OK_NO_DATA: Final = "OK_NO_DATA"
+UPDATE_ERROR: Final = "ERROR"
 
-XML_ATTR_HREF = "@href"
-XML_ATTR_TERM = "@term"
+XML_ATTR_HREF: Final = "@href"
+XML_ATTR_TERM: Final = "@term"
 
-XML_CDATA = "#text"
+XML_CDATA: Final = "#text"
 
-XML_TAG_AUTHOR = "author"
-XML_TAG_CATEGORY = "category"
-XML_TAG_CHANNEL = "channel"
-XML_TAG_CONTENT = "content"
-XML_TAG_CONTRIBUTOR = "contributor"
-XML_TAG_COPYRIGHT = "copyright"
-XML_TAG_DC_DATE = "dc:date"
-XML_TAG_DESCRIPTION = "description"
-XML_TAG_DOCS = "docs"
-XML_TAG_ENTRY = "entry"
-XML_TAG_FEED = "feed"
-XML_TAG_GDACS_BBOX = "gdacs:bbox"
-XML_TAG_GENERATOR = "generator"
-XML_TAG_GEO_LAT = "geo:lat"
-XML_TAG_GEO_LONG = "geo:long"
-XML_TAG_GEO_POINT = "geo:Point"
-XML_TAG_GEORSS_POINT = "georss:point"
-XML_TAG_GEORSS_POLYGON = "georss:polygon"
-XML_TAG_GEORSS_WHERE = "georss:where"
-XML_TAG_GML_EXTERIOR = "gml:exterior"
-XML_TAG_GML_LINEAR_RING = "gml:LinearRing"
-XML_TAG_GML_POINT = "gml:Point"
-XML_TAG_GML_POLYGON = "gml:Polygon"
-XML_TAG_GML_POS = "gml:pos"
-XML_TAG_GML_POS_LIST = "gml:posList"
-XML_TAG_GUID = "guid"
-XML_TAG_HEIGHT = "height"
-XML_TAG_ID = "id"
-XML_TAG_IMAGE = "image"
-XML_TAG_ITEM = "item"
-XML_TAG_LANGUAGE = "language"
-XML_TAG_LAST_BUILD_DATE = "lastBuildDate"
-XML_TAG_LINK = "link"
-XML_TAG_MANAGING_EDITOR = "managingEditor"
-XML_TAG_NAME = "name"
-XML_TAG_PUB_DATE = "pubDate"
-XML_TAG_PUBLISHED = "published"
-XML_TAG_RIGHTS = "rights"
-XML_TAG_RSS = "rss"
-XML_TAG_SOURCE = "source"
-XML_TAG_SUBTITLE = "subtitle"
-XML_TAG_SUMMARY = "summary"
-XML_TAG_TITLE = "title"
-XML_TAG_TTL = "ttl"
-XML_TAG_UPDATED = "updated"
-XML_TAG_URL = "url"
-XML_TAG_WIDTH = "width"
+XML_TAG_AUTHOR: Final = "author"
+XML_TAG_CATEGORY: Final = "category"
+XML_TAG_CHANNEL: Final = "channel"
+XML_TAG_CONTENT: Final = "content"
+XML_TAG_CONTRIBUTOR: Final = "contributor"
+XML_TAG_COPYRIGHT: Final = "copyright"
+XML_TAG_DC_DATE: Final = "dc:date"
+XML_TAG_DESCRIPTION: Final = "description"
+XML_TAG_DOCS: Final = "docs"
+XML_TAG_ENTRY: Final = "entry"
+XML_TAG_FEED: Final = "feed"
+XML_TAG_GDACS_BBOX: Final = "gdacs:bbox"
+XML_TAG_GENERATOR: Final = "generator"
+XML_TAG_GEO_LAT: Final = "geo:lat"
+XML_TAG_GEO_LONG: Final = "geo:long"
+XML_TAG_GEO_POINT: Final = "geo:Point"
+XML_TAG_GEORSS_POINT: Final = "georss:point"
+XML_TAG_GEORSS_POLYGON: Final = "georss:polygon"
+XML_TAG_GEORSS_WHERE: Final = "georss:where"
+XML_TAG_GML_EXTERIOR: Final = "gml:exterior"
+XML_TAG_GML_LINEAR_RING: Final = "gml:LinearRing"
+XML_TAG_GML_POINT: Final = "gml:Point"
+XML_TAG_GML_POLYGON: Final = "gml:Polygon"
+XML_TAG_GML_POS: Final = "gml:pos"
+XML_TAG_GML_POS_LIST: Final = "gml:posList"
+XML_TAG_GUID: Final = "guid"
+XML_TAG_HEIGHT: Final = "height"
+XML_TAG_ID: Final = "id"
+XML_TAG_IMAGE: Final = "image"
+XML_TAG_ITEM: Final = "item"
+XML_TAG_LANGUAGE: Final = "language"
+XML_TAG_LAST_BUILD_DATE: Final = "lastBuildDate"
+XML_TAG_LINK: Final = "link"
+XML_TAG_MANAGING_EDITOR: Final = "managingEditor"
+XML_TAG_NAME: Final = "name"
+XML_TAG_PUB_DATE: Final = "pubDate"
+XML_TAG_PUBLISHED: Final = "published"
+XML_TAG_RIGHTS: Final = "rights"
+XML_TAG_RSS: Final = "rss"
+XML_TAG_SOURCE: Final = "source"
+XML_TAG_SUBTITLE: Final = "subtitle"
+XML_TAG_SUMMARY: Final = "summary"
+XML_TAG_TITLE: Final = "title"
+XML_TAG_TTL: Final = "ttl"
+XML_TAG_UPDATED: Final = "updated"
+XML_TAG_URL: Final = "url"
+XML_TAG_WIDTH: Final = "width"

aio_georss_client/feed.py

@@ -1,10 +1,11 @@
 """GeoRSS Feed."""
-import asyncio
+from __future__ import annotations
+
 import codecs
 import logging
 from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import Dict, Generic, List, Optional, Tuple, TypeVar
+from typing import Generic, TypeVar
 
 import aiohttp
 from aiohttp import ClientSession, client_exceptions
@@ -32,18 +33,18 @@ class GeoRssFeed(Generic[T_FEED_ENTRY], ABC):
     def __init__(
         self,
         websession: ClientSession,
-        home_coordinates: Tuple[float, float],
+        home_coordinates: tuple[float, float],
         url: str,
-        filter_radius: float = None,
-        filter_categories: List[str] = None,
+        filter_radius: float | None = None,
+        filter_categories: list[str] | None = None,
     ):
         """Initialise this service."""
         self._websession = websession
-        self._home_coordinates = home_coordinates
-        self._filter_radius = filter_radius
-        self._filter_categories = filter_categories
-        self._url = url
-        self._last_timestamp = None
+        self._home_coordinates: tuple[float, float] = home_coordinates
+        self._filter_radius: float | None = filter_radius
+        self._filter_categories: list[str] | None = filter_categories
+        self._url: str = url
+        self._last_timestamp: datetime | None = None
 
     def __repr__(self):
         """Return string representation of this feed."""
@@ -58,9 +59,9 @@ def __repr__(self):
     @abstractmethod
     def _new_entry(
         self,
-        home_coordinates: Tuple[float, float],
+        home_coordinates: tuple[float, float],
         rss_entry: FeedItem,
-        global_data: Dict,
+        global_data: dict,
     ) -> T_FEED_ENTRY:
         """Generate a new entry."""
         pass
@@ -73,7 +74,7 @@ def _additional_namespaces(self):
         """Provide additional namespaces, relevant for this feed."""
         pass
 
-    async def update(self) -> Tuple[str, Optional[List[T_FEED_ENTRY]]]:
+    async def update(self) -> tuple[str, list[T_FEED_ENTRY] | None]:
         """Update from external source and return filtered entries."""
         status, rss_data = await self._fetch()
         if status == UPDATE_OK:
@@ -101,7 +102,7 @@ async def update(self) -> Tuple[str, Optional[List[T_FEED_ENTRY]]]:
 
     async def _fetch(
         self, method: str = "GET", headers=None, params=None
-    ) -> Tuple[str, Optional[Feed]]:
+    ) -> tuple[str, Feed | None]:
         """Fetch GeoRSS data from external source."""
         try:
             timeout = aiohttp.ClientTimeout(total=self._client_session_timeout())
@@ -133,7 +134,7 @@ async def _fetch(
                 client_error,
             )
             return UPDATE_ERROR, None
-        except asyncio.TimeoutError:
+        except TimeoutError:
             _LOGGER.warning(
                 "Requesting data from %s failed with timeout error", self._url
             )
@@ -149,7 +150,7 @@ async def _read_response(self, response):
             return await response.text()
         return None
 
-    def _filter_entries(self, entries: List[T_FEED_ENTRY]):
+    def _filter_entries(self, entries: list[T_FEED_ENTRY]):
         """Filter the provided entries."""
         filtered_entries = entries
         _LOGGER.debug("Entries before filtering %s", filtered_entries)
@@ -183,7 +184,7 @@ def _filter_entries(self, entries: List[T_FEED_ENTRY]):
         _LOGGER.debug("Entries after filtering %s", filtered_entries)
         return filtered_entries
 
-    def _extract_from_feed(self, feed: Feed) -> Dict:
+    def _extract_from_feed(self, feed: Feed) -> dict:
         """Extract global metadata from feed."""
         global_data = {}
         author = feed.author
@@ -192,8 +193,8 @@ def _extract_from_feed(self, feed: Feed) -> Dict:
         return global_data
 
     def _extract_last_timestamp(
-        self, feed_entries: List[T_FEED_ENTRY]
-    ) -> Optional[datetime]:
+        self, feed_entries: list[T_FEED_ENTRY]
+    ) -> datetime | None:
         """Determine latest (newest) entry from the filtered feed."""
         if feed_entries:
             dates = sorted(
@@ -207,6 +208,6 @@ def _extract_last_timestamp(
         return None
 
     @property
-    def last_timestamp(self) -> Optional[datetime]:
+    def last_timestamp(self) -> datetime | None:
         """Return the last timestamp extracted from this feed."""
         return self._last_timestamp

aio_georss_client/feed_entry.py

@@ -1,9 +1,10 @@
 """Feed Entry."""
+from __future__ import annotations
+
 import logging
 import re
 from abc import ABC, abstractmethod
 from datetime import datetime
-from typing import List, Optional, Tuple, Type
 
 from .consts import CUSTOM_ATTRIBUTE
 from .geo_rss_distance_helper import GeoRssDistanceHelper
@@ -18,22 +19,22 @@
 class FeedEntry(ABC):
     """Feed entry base class."""
 
-    def __init__(self, home_coordinates: Tuple[float, float], rss_entry: FeedItem):
+    def __init__(self, home_coordinates: tuple[float, float], rss_entry: FeedItem):
         """Initialise this feed entry."""
         self._home_coordinates = home_coordinates
         self._rss_entry = rss_entry
 
     def __repr__(self):
         """Return string representation of this entry."""
-        return "<{}(id={})>".format(self.__class__.__name__, self.external_id)
+        return f"<{self.__class__.__name__}(id={self.external_id})>"
 
     @property
-    def features(self) -> List[Type[Geometry]]:
+    def features(self) -> list[type[Geometry]]:
         """Return the list of geometry types that this feed entry supports."""
         return DEFAULT_FEATURES
 
     @property
-    def geometries(self) -> Optional[List[Geometry]]:
+    def geometries(self) -> list[Geometry] | None:
         """Return all geometries of this entry."""
         if self._rss_entry:
             # Return all geometries that are of type defined in features.
@@ -43,7 +44,7 @@ def geometries(self) -> Optional[List[Geometry]]:
         return None
 
     @property
-    def coordinates(self) -> Optional[Tuple[float, float]]:
+    def coordinates(self) -> tuple[float, float] | None:
         """Return the best coordinates (latitude, longitude) of this entry."""
         # This looks for the first point in the list of geometries. If there
         # is no point then return the first entry.
@@ -56,7 +57,7 @@ def coordinates(self) -> Optional[Tuple[float, float]]:
         return None
 
     @property
-    def external_id(self) -> Optional[str]:
+    def external_id(self) -> str | None:
         """Return the external id of this entry."""
         if self._rss_entry:
             external_id = self._rss_entry.guid
@@ -68,7 +69,7 @@ def external_id(self) -> Optional[str]:
             return external_id
         return None
 
-    def _search_in_external_id(self, regexp) -> Optional[str]:
+    def _search_in_external_id(self, regexp) -> str | None:
         """Find a sub-string in the entry's external id."""
         if self.external_id:
             match = re.search(regexp, self.external_id)
@@ -77,7 +78,7 @@ def _search_in_external_id(self, regexp) -> Optional[str]:
         return None
 
     @property
-    def title(self) -> Optional[str]:
+    def title(self) -> str | None:
         """Return the title of this entry."""
         if self._rss_entry:
             return self._rss_entry.title
@@ -92,7 +93,7 @@ def _search_in_title(self, regexp):
         return None
 
     @property
-    def category(self) -> Optional[str]:
+    def category(self) -> str | None:
         """Return the category of this entry."""
         if (
             self._rss_entry
@@ -105,7 +106,7 @@ def category(self) -> Optional[str]:
 
     @property
     @abstractmethod
-    def attribution(self) -> Optional[str]:
+    def attribution(self) -> str | None:
         """Return the attribution of this entry."""
         return None
 
@@ -126,21 +127,21 @@ def distance_to_home(self) -> float:
         return distance
 
     @property
-    def description(self) -> Optional[str]:
+    def description(self) -> str | None:
         """Return the description of this entry."""
         if self._rss_entry and self._rss_entry.description:
             return self._rss_entry.description
         return None
 
     @property
-    def published(self) -> Optional[datetime]:
+    def published(self) -> datetime | None:
         """Return the published date of this entry."""
         if self._rss_entry:
             return self._rss_entry.published_date
         return None
 
     @property
-    def updated(self) -> Optional[datetime]:
+    def updated(self) -> datetime | None:
         """Return the updated date of this entry."""
         if self._rss_entry:
             return self._rss_entry.updated_date