aboutsummaryrefslogtreecommitdiff
path: root/repos
diff options
context:
space:
mode:
Diffstat (limited to 'repos')
-rw-r--r--repos/__init__.py2
-rw-r--r--repos/alpine_linux.py15
-rw-r--r--repos/base.py55
3 files changed, 59 insertions, 13 deletions
diff --git a/repos/__init__.py b/repos/__init__.py
index 0f71dd1..1dea3a1 100644
--- a/repos/__init__.py
+++ b/repos/__init__.py
@@ -1,4 +1,4 @@
-from typing import Mapping, List, Any
+from typing import Mapping, List
from . import alpine_linux
from .base import Repository
diff --git a/repos/alpine_linux.py b/repos/alpine_linux.py
index 64f159a..7c646cc 100644
--- a/repos/alpine_linux.py
+++ b/repos/alpine_linux.py
@@ -1,9 +1,10 @@
from io import TextIOWrapper
from pathlib import Path
+import re
import tarfile
from typing import Mapping, TextIO
-from .base import Repository
+from .base import Repository, Version
__all__ = [
'stable_main_x86_64',
@@ -13,7 +14,9 @@ __all__ = [
'edge_testing_x86_64',
]
-def parse_apkindex(apkindex: TextIO) -> Mapping[str, str]:
+PACKAGE_REVISION_INFO = re.compile(r'-r\d+$')
+
+def parse_apkindex(apkindex: TextIO) -> Mapping[str, Version]:
result = dict()
current_package = None
current_version = None
@@ -37,14 +40,16 @@ def parse_apkindex(apkindex: TextIO) -> Mapping[str, str]:
elif line_type == 'P':
current_package = line_data
elif line_type == 'V':
- current_version = line_data
+ version = line_data
+ clean_version = PACKAGE_REVISION_INFO.sub('', version)
+ current_version = Version(version, clean_version)
elif line_type in ignore_lines:
pass
else:
raise ValueError('unknown line type: ' + line_type + ' in line ' + repr(line))
return result
-def parse_cached(cached: Path) -> Mapping[str, str]:
+def parse_cached(cached: Path) -> Mapping[str, Version]:
apkindex = tarfile.open(cached)
for archive_member in apkindex.getmembers():
if archive_member.name == 'APKINDEX':
@@ -52,7 +57,7 @@ def parse_cached(cached: Path) -> Mapping[str, str]:
apkindex_file = TextIOWrapper(apkindex_file)
return parse_apkindex(apkindex_file)
-def build_repo(name: str, url_path: str):
+def build_repo(name: str, url_path: str) -> Repository:
url = f'http://dl-cdn.alpinelinux.org/alpine/{url_path}/APKINDEX.tar.gz'
return Repository(
family='Alpine Linux',
diff --git a/repos/base.py b/repos/base.py
index 66ecf2d..220a30f 100644
--- a/repos/base.py
+++ b/repos/base.py
@@ -1,26 +1,66 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, asdict as dataclass_asdict
import datetime
+from functools import total_ordering
import gzip
import json
import os
from pathlib import Path
import re
-from typing import Callable, Mapping
+from typing import Any, Callable, Mapping
import requests
+import semver
+
+__all__ = [
+ 'Repository',
+ 'Version',
+]
HTTP_DATE = '%a, %d %b %Y %H:%M:%S GMT'
-SLUGIFY = re.compile('\W+')
+SLUGIFY = re.compile(r'\W+')
def slug(text: str) -> str:
return SLUGIFY.sub('-', text.lower()).strip('-')
+@total_ordering
+@dataclass()
+class Version:
+ original: str
+ clean: str
+
+ def __str__(self) -> str:
+ return self.original
+
+ def __lt__(self, other: Any):
+ if not isinstance(other, Version):
+ return NotImplemented
+ if semver.VersionInfo.isvalid(self.clean) and semver.VersionInfo.isvalid(other.clean):
+ return semver.compare(self.clean, other.clean) < 0
+ return self.original < other.original
+
+class JSONEncoder(json.JSONEncoder):
+ def default(self, o: Any) -> Any:
+ if isinstance(o, Version):
+ return dataclass_asdict(o)
+ return super().default(o)
+
+class JSONDecoder(json.JSONDecoder):
+ @staticmethod
+ def object_hook(o: dict) -> Any:
+ if o.keys() == {'original', 'clean'}:
+ return Version(**o)
+ return o
+
+ def __init__(self):
+ super().__init__(object_hook=self.object_hook)
+
+
@dataclass()
class Repository:
family: str
repo: str
index_url: str
- parse: Callable[[Path], Mapping[str, str]]
+ parse: Callable[[Path], Mapping[str, Version]]
def _full_name(self):
return f'{self.family} {self.repo}'
@@ -31,7 +71,7 @@ class Repository:
def _cache_file(self, name: str) -> Path:
return self._cache_dir() / name
- def get_versions(self) -> Mapping[str, str]:
+ def get_versions(self) -> Mapping[str, Version]:
self._cache_dir().mkdir(parents=True, exist_ok=True)
downloaded_file = self._cache_file('downloaded')
if downloaded_file.exists():
@@ -57,10 +97,11 @@ class Repository:
set_mtime = datetime.datetime.strptime(set_mtime, HTTP_DATE)
os.utime(downloaded_file, (datetime.datetime.now().timestamp(), set_mtime.timestamp()))
+ if response.status_code != requests.codes.not_modified or not parsed_file.exists():
parsed_data = self.parse(downloaded_file)
with gzip.open(parsed_file, 'wt') as f:
- json.dump(parsed_data, f)
+ json.dump(parsed_data, f, cls=JSONEncoder)
return parsed_data
else:
with gzip.open(parsed_file, 'rt') as f:
- return json.load(f)
+ return json.load(f, cls=JSONDecoder)