|
3 | 3 | import dataclasses |
4 | 4 | import json |
5 | 5 | from pathlib import Path |
6 | | -from typing import Optional, Union |
| 6 | +from typing import List, Optional, Union |
7 | 7 |
|
8 | 8 | import requests |
9 | 9 |
|
@@ -102,3 +102,38 @@ def from_ogc_api_record(cls, src: Union[dict, str, Path]) -> Algorithm: |
102 | 102 | description=properties.get("description"), |
103 | 103 | udp_link=udp_link, |
104 | 104 | ) |
| 105 | + |
| 106 | + |
| 107 | +class GithubAlgorithmRepository: |
| 108 | + """ |
| 109 | + GitHub based algorithm repository. |
| 110 | + """ |
| 111 | + |
| 112 | + # TODO: caching |
| 113 | + |
| 114 | + def __init__(self, owner: str, repo: str, folder: str = "", branch: str = "main"): |
| 115 | + self.owner = owner |
| 116 | + self.repo = repo |
| 117 | + self.folder = folder |
| 118 | + self.branch = branch |
| 119 | + self._session = requests.Session() |
| 120 | + |
| 121 | + def _list_files(self): |
| 122 | + url = f"https://api.github.com/repos/{self.owner}/{self.repo}/contents/{self.folder}".strip("/") |
| 123 | + resp = self._session.get(url, headers={"Accept": "application/vnd.github.object+json"}) |
| 124 | + resp.raise_for_status() |
| 125 | + listing = resp.json() |
| 126 | + assert listing["type"] == "dir" |
| 127 | + for item in listing["entries"]: |
| 128 | + if item["type"] == "file": |
| 129 | + yield item |
| 130 | + |
| 131 | + def list_algorithms(self) -> List[str]: |
| 132 | + # TODO: method to list names vs method to list parsed Algorithm objects? |
| 133 | + return [item["name"] for item in self._list_files()] |
| 134 | + |
| 135 | + def get_algorithm(self, name: str) -> Algorithm: |
| 136 | + # TODO: get url from listing from API request, instead of hardcoding this raw url? |
| 137 | + url = f"https://raw.githubusercontent.com/{self.owner}/{self.repo}/{self.branch}/{self.folder}/{name}" |
| 138 | + # TODO: how to make sure GitHub URL is requested with additional headers? |
| 139 | + return Algorithm.from_ogc_api_record(url) |
0 commit comments