161 lines
4.4 KiB
Python
161 lines
4.4 KiB
Python
import json
|
|
from enum import Enum, auto
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
from urllib.parse import ParseResult, urlparse
|
|
|
|
from .fileutils import PathType, to_path, write_json_file
|
|
|
|
Url = ParseResult
|
|
|
|
|
|
class LockedVersion:
|
|
def __init__(
|
|
self, url: Url, rev: str, sha256: str, submodules: bool = False
|
|
) -> None:
|
|
self.url = url
|
|
self.rev = rev
|
|
self.sha256 = sha256
|
|
self.submodules = submodules
|
|
|
|
def __eq__(self, other: Any) -> bool:
|
|
if type(other) is type(self):
|
|
return self.__dict__ == other.__dict__
|
|
return False
|
|
|
|
def as_json(self) -> Dict[str, Any]:
|
|
d = dict(
|
|
url=self.url.geturl(), rev=self.rev, sha256=self.sha256
|
|
) # type: Dict[str, Any]
|
|
if self.submodules:
|
|
d["submodules"] = self.submodules
|
|
return d
|
|
|
|
|
|
class RepoType(Enum):
|
|
GITHUB = auto()
|
|
GITLAB = auto()
|
|
GIT = auto()
|
|
|
|
@staticmethod
|
|
def from_repo(repo: "Repo", type_: Optional[str]) -> "RepoType":
|
|
if repo.submodules:
|
|
return RepoType.GIT
|
|
if repo.url.hostname == "github.com":
|
|
return RepoType.GITHUB
|
|
if repo.url.hostname == "gitlab.com" or type_ == "gitlab":
|
|
return RepoType.GITLAB
|
|
else:
|
|
return RepoType.GIT
|
|
|
|
|
|
class Repo:
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
url: Url,
|
|
submodules: bool,
|
|
supplied_type: Optional[str],
|
|
file_: Optional[str],
|
|
branch: Optional[str],
|
|
locked_version: Optional[LockedVersion],
|
|
) -> None:
|
|
self.name = name
|
|
self.url = url
|
|
self.submodules = submodules
|
|
if file_ is None:
|
|
self.file = "default.nix"
|
|
else:
|
|
self.file = file_
|
|
self.branch = branch
|
|
self.locked_version = None
|
|
|
|
if (
|
|
locked_version is not None
|
|
and locked_version.url != url.geturl()
|
|
and locked_version.submodules == submodules
|
|
):
|
|
self.locked_version = locked_version
|
|
|
|
self.supplied_type = supplied_type
|
|
self.computed_type = RepoType.from_repo(self, supplied_type)
|
|
|
|
@property
|
|
def type(self) -> RepoType:
|
|
return self.computed_type
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<{self.__class__.__name__} {self.name}>"
|
|
|
|
def as_json(self) -> Dict[str, Any]:
|
|
d = dict(url=self.url.geturl()) # type: Dict[str, Any]
|
|
|
|
if self.submodules:
|
|
d["submodules"] = self.submodules
|
|
|
|
if self.supplied_type is not None:
|
|
d["type"] = self.supplied_type
|
|
|
|
if self.file is not None and self.file != "default.nix":
|
|
d["file"] = self.file
|
|
|
|
return d
|
|
|
|
|
|
class Manifest:
|
|
def __init__(self, repos: List[Repo]) -> None:
|
|
self.repos = repos
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<{self.__class__.__name__} {repr(self.repos)}>"
|
|
|
|
|
|
def _load_locked_versions(path: PathType) -> Dict[str, LockedVersion]:
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
|
|
locked_versions = {}
|
|
|
|
for name, repo in data["repos"].items():
|
|
url = urlparse(repo["url"])
|
|
rev = repo["rev"]
|
|
sha256 = repo["sha256"]
|
|
submodules = repo.get("submodules", False)
|
|
locked_versions[name] = LockedVersion(url, rev, sha256, submodules)
|
|
|
|
return locked_versions
|
|
|
|
|
|
def load_locked_versions(path: Path) -> Dict[str, LockedVersion]:
|
|
if path.exists():
|
|
return _load_locked_versions(path)
|
|
else:
|
|
return {}
|
|
|
|
|
|
def update_lock_file(repos: List[Repo], path: Path) -> None:
|
|
locked_repos = {}
|
|
for repo in repos:
|
|
if repo.locked_version:
|
|
locked_repos[repo.name] = repo.locked_version.as_json()
|
|
|
|
write_json_file(dict(repos=locked_repos), path)
|
|
|
|
|
|
def load_manifest(manifest_path: PathType, lock_path: PathType) -> Manifest:
|
|
locked_versions = load_locked_versions(to_path(lock_path))
|
|
|
|
with open(manifest_path) as f:
|
|
data = json.load(f)
|
|
|
|
repos = []
|
|
for name, repo in data["repos"].items():
|
|
url = urlparse(repo["url"])
|
|
submodules = repo.get("submodules", False)
|
|
branch_ = repo.get("branch")
|
|
file_ = repo.get("file", "default.nix")
|
|
type_ = repo.get("type", None)
|
|
locked_version = locked_versions.get(name)
|
|
repos.append(Repo(name, url, submodules, type_, file_, branch_, locked_version))
|
|
|
|
return Manifest(repos)
|