diff --git a/README.md b/README.md index 19355c0..b0dea59 100644 --- a/README.md +++ b/README.md @@ -44,33 +44,34 @@ from pydantic_pkgr import * apt, brew, pip, npm, env = AptProvider(), BrewProvider(), PipProvider(), NpmProvider(), EnvProvider() dependencies = [ - Binary(name='curl', providers=[env, apt, brew]), - Binary(name='wget', providers=[env, apt, brew]), - Binary(name='yt-dlp', providers=[env, pip, apt, brew]), - Binary(name='playwright', providers=[env, pip, npm]), - Binary(name='puppeteer', providers=[env, npm]), + Binary(name='curl', binproviders=[env, apt, brew]), + Binary(name='wget', binproviders=[env, apt, brew]), + Binary(name='yt-dlp', binproviders=[env, pip, apt, brew]), + Binary(name='playwright', binproviders=[env, pip, npm]), + Binary(name='puppeteer', binproviders=[env, npm]), ] for binary in dependencies: binary = binary.load_or_install() - print(binary.abspath, binary.version, binary.provider, binary.is_valid) - # Path('/usr/bin/curl') SemVer('7.81.0') 'apt' True + print(binary.abspath, binary.version, binary.binprovider, binary.is_valid) + # Path('/usr/bin/curl') SemVer('7.81.0') AptProvider() True binary.exec(cmd=['--version']) # curl 7.81.0 (x86_64-apple-darwin23.0) libcurl/7.81.0 ... ``` ```python +from pydantic import InstanceOf from pydantic_pkgr import Binary, BinProvider, BrewProvider, EnvProvider # you can also define binaries as classes, making them usable for type checking class CurlBinary(Binary): name: str = 'curl' - providers: list[BinProvider] = [BrewProvider(), EnvProvider()] + binproviders: List[InstanceOf[BinProvider]] = [BrewProvider(), EnvProvider()] curl = CurlBinary().install() -assert isinstance(curl, CurlBinary) # CurlBinary is a unique type you can use in annotations now -print(curl.abspath, curl.version, curl.provider, curl.is_valid) # Path('/opt/homebrew/bin/curl') SemVer('8.4.0') 'brew' True -curl.exec(cmd=['--version']) # curl 8.4.0 (x86_64-apple-darwin23.0) libcurl/8.4.0 ... +assert isinstance(curl, CurlBinary) # CurlBinary is a unique type you can use in annotations now +print(curl.abspath, curl.version, curl.binprovider, curl.is_valid) # Path('/opt/homebrew/bin/curl') SemVer('8.4.0') BrewProvider() True +curl.exec(cmd=['--version']) # curl 8.4.0 (x86_64-apple-darwin23.0) libcurl/8.4.0 ... ``` ```python @@ -150,7 +151,7 @@ curl.exec(['--version']) # curl 7.81.0 (x86_64-pc-linux-gnu) libcur ### Example: Finding/Installing django with pip (w/ customized binpath resolution behavior) pip = PipProvider( - abspath_provider={'*': lambda bin_name, **context: inspect.getfile(bin_name)}, # use python inspect to get path instead of os.which + abspath_handler={'*': lambda bin_name, **context: inspect.getfile(bin_name)}, # use python inspect to get path instead of os.which ) django_bin = pip.load_or_install(bin_name='django') print(django_bin.abspath) # Path('/usr/lib/python3.10/site-packages/django/__init__.py') @@ -164,7 +165,7 @@ It can define one or more `BinProvider`s that it supports, along with overrides `Binary`s implement the following interface: - `load()`, `install()`, `load_or_install()` `->` `Binary` -- `provider: BinProviderName` (`BinProviderName == str`) +- `binprovider: InstanceOf[BinProvider]` - `abspath: Path` - `abspaths: List[Path]` - `version: SemVer` @@ -177,7 +178,7 @@ class YtdlpBinary(Binary): name: BinName = 'ytdlp' description: str = 'YT-DLP (Replacement for YouTube-DL) Media Downloader' - providers_supported: List[BinProvider] = [EnvProvider(), PipProvider(), AptProvider(), BrewProvider()] + binproviders_supported: List[BinProvider] = [EnvProvider(), PipProvider(), AptProvider(), BrewProvider()] # customize installed package names for specific package managers provider_overrides: Dict[BinProviderName, ProviderLookupDict] = { @@ -187,7 +188,7 @@ class YtdlpBinary(Binary): } ytdlp = YtdlpBinary().load_or_install() -print(ytdlp.provider) # 'brew' +print(ytdlp.binprovider) # BrewProvider(...) print(ytdlp.abspath) # Path('/opt/homebrew/bin/yt-dlp') print(ytdlp.abspaths) # [Path('/opt/homebrew/bin/yt-dlp'), Path('/usr/local/bin/yt-dlp')] print(ytdlp.version) # SemVer('2024.4.9') @@ -201,7 +202,7 @@ from pydantic_pkgr import BinProvider, Binary, BinProviderName, BinName, Provide class DockerBinary(Binary): name: BinName = 'docker' - providers_supported: List[BinProvider] = [EnvProvider(), AptProvider()] + binproviders_supported: List[BinProvider] = [EnvProvider(), AptProvider()] provider_overrides: Dict[BinProviderName, ProviderLookupDict] = { 'env': { @@ -219,7 +220,7 @@ class DockerBinary(Binary): } docker = DockerBinary().load_or_install() -print(docker.provider) # 'env' +print(docker.binprovider) # EnvProvider() print(docker.abspath) # Path('/usr/local/bin/podman') print(docker.abspaths) # [Path('/usr/local/bin/podman'), Path('/opt/homebrew/bin/podman')] print(docker.version) # SemVer('6.0.2') @@ -229,7 +230,7 @@ print(docker.is_valid) # True # e.g. if you want to force the abspath to be at a specific path: custom_docker = DockerBinary(abspath='~/custom/bin/podman').load() print(custom_docker.name) # 'docker' -print(custom_docker.provider) # 'env' +print(custom_docker.binprovider) # EnvProvider() print(custom_docker.abspath) # Path('/Users/example/custom/bin/podman') print(custom_docker.version) # SemVer('5.0.2') print(custom_docker.is_valid) # True @@ -282,15 +283,16 @@ pip install django-pydantic-field Example Django `models.py` showing how to store `Binary` and `BinProvider` instances in DB fields: ```python +from typing import List from django.db import models -from django_pydantic_field import SchemaField - +from pydantic import InstanceOf from pydantic_pkgr import BinProvider, Binary, SemVer +from django_pydantic_field import SchemaField class InstalledBinary(models.Model): name = models.CharField(max_length=63) binary: Binary = SchemaField() - providers: list[BinProvider] = SchemaField(default=[]) + binproviders: List[InstanceOf[BinProvider]] = SchemaField(default=[]) version: SemVer = SchemaField(default=(0,0,1)) ``` @@ -303,7 +305,7 @@ curl = Binary(name='curl').load() obj = InstalledBinary( name='curl', binary=curl, # store Binary/BinProvider/SemVer values directly in fields - providers=[env], # no need for manual JSON serialization / schema checking + binproviders=[env], # no need for manual JSON serialization / schema checking min_version=SemVer('6.5.0'), ) obj.save() @@ -454,7 +456,7 @@ class CargoProvider(BinProvider): cargo = CargoProvider() rg = cargo.install(bin_name='ripgrep') -print(rg.provider) # 'cargo' +print(rg.binprovider) # CargoProvider() print(rg.version) # SemVer(14, 1, 0) ``` diff --git a/pydantic_pkgr/binary.py b/pydantic_pkgr/binary.py index 2f781a6..82f54bd 100644 --- a/pydantic_pkgr/binary.py +++ b/pydantic_pkgr/binary.py @@ -13,7 +13,7 @@ from pydantic_core import ValidationError -from pydantic import BaseModel, Field, model_validator, computed_field, field_validator, validate_call, field_serializer, ConfigDict +from pydantic import BaseModel, Field, model_validator, computed_field, field_validator, validate_call, field_serializer, ConfigDict, InstanceOf from .semver import SemVer from .binprovider import ( @@ -44,10 +44,10 @@ class Binary(ShallowBinary): name: BinName = '' description: str = '' - providers_supported: List[BinProvider] = Field(default=[DEFAULT_PROVIDER], alias='providers') + binproviders_supported: List[InstanceOf[BinProvider]] = Field(default=[DEFAULT_PROVIDER], alias='binproviders') provider_overrides: Dict[BinProviderName, ProviderLookupDict] = Field(default={}, alias='overrides') - loaded_provider: Optional[BinProviderName] = Field(default=None, alias='provider') + loaded_binprovider: Optional[InstanceOf[BinProvider]] = Field(default=None, alias='binprovider') loaded_abspath: Optional[HostBinPath] = Field(default=None, alias='abspath') loaded_version: Optional[SemVer] = Field(default=None, alias='version') @@ -62,15 +62,15 @@ def validate(self): # assert self.name, 'Binary.name must not be empty' self.description = self.description or self.name - assert self.providers_supported, f'No providers were given for package {self.name}' + assert self.binproviders_supported, f'No providers were given for package {self.name}' # pull in any overrides from the binproviders - for provider in self.providers_supported: - overrides_by_provider = provider.get_providers_for_bin(self.name) - if overrides_by_provider: - self.provider_overrides[provider.name] = { - **overrides_by_provider, - **self.provider_overrides.get(provider.name, {}), + for binprovider in self.binproviders_supported: + overrides_by_handler = binprovider.get_handlers_for_bin(self.name) + if overrides_by_handler: + self.provider_overrides[binprovider.name] = { + **overrides_by_handler, + **self.provider_overrides.get(binprovider.name, {}), } return self @@ -85,11 +85,11 @@ def parse_version(cls, value: Any) -> Optional[SemVer]: @field_serializer('provider_overrides', when_used='json') def serialize_overrides(self, provider_overrides: Dict[BinProviderName, ProviderLookupDict]) -> Dict[BinProviderName, Dict[str, str]]: return { - provider_name: { + binprovider_name: { key: str(val) for key, val in overrides.items() } - for provider_name, overrides in provider_overrides.items() + for binprovider_name, overrides in provider_overrides.items() } @computed_field @@ -99,15 +99,15 @@ def loaded_abspaths(self) -> Dict[BinProviderName, List[HostBinPath]]: # binary has not been loaded yet return {} - all_bin_abspaths = {self.loaded_provider: [self.loaded_abspath]} if self.loaded_provider else {} - for provider in self.providers_supported: - if not provider.PATH: - # print('skipping provider', provider.name, provider.PATH) + all_bin_abspaths = {self.loaded_binprovider.name: [self.loaded_abspath]} if self.loaded_binprovider else {} + for binprovider in self.binproviders_supported: + if not binprovider.PATH: + # print('skipping provider', binprovider.name, binprovider.PATH) continue - for bin_abspath in bin_abspaths(self.name, PATH=provider.PATH): - existing = all_bin_abspaths.get(provider.name, []) + for bin_abspath in bin_abspaths(self.name, PATH=binprovider.PATH): + existing = all_bin_abspaths.get(binprovider.name, []) if bin_abspath not in existing: - all_bin_abspaths[provider.name] = [ + all_bin_abspaths[binprovider.name] = [ *existing, bin_abspath, ] @@ -121,33 +121,27 @@ def loaded_bin_dirs(self) -> Dict[BinProviderName, BinDirPath]: provider_name: ':'.join([str(bin_abspath.parent) for bin_abspath in bin_abspaths]) for provider_name, bin_abspaths in self.loaded_abspaths.items() } - - @computed_field - @property - def PROVIDER(self) -> Optional[BinProvider]: - if not self.loaded_provider: - return None - for provider in self.providers_supported: - if provider.name == self.loaded_provider: - return provider - raise Exception(f'Binary {self.name} was loaded using {self.loaded_provider} BinProvider, but {self.loaded_provider} was not found in self.providers_supported={self.providers_supported}') - @validate_call def install(self) -> Self: assert self.name, f'No binary name was provided! {self}' - if not self.providers_supported: + if not self.binproviders_supported: return self - outer_exc = Exception(f'None of the configured providers [{", ".join(p.name for p in self.providers_supported)}] were able to install binary: {self.name}') + outer_exc = Exception(f'None of the configured providers [{", ".join(p.name for p in self.binproviders_supported)}] were able to install binary: {self.name}') inner_exc = Exception('No providers were available') - for provider in self.providers_supported: + for binprovider in self.binproviders_supported: try: - installed_bin = provider.install(self.name, overrides=self.provider_overrides.get(provider.name)) + installed_bin = binprovider.install(self.name, overrides=self.provider_overrides.get(binprovider.name)) if installed_bin: # print('INSTALLED', self.name, installed_bin) - return self.__class__.model_validate({**self.model_dump(), **installed_bin.model_dump(exclude=('providers_supported',))}) + return self.__class__.model_validate({ + **self.model_dump(), + **installed_bin.model_dump(exclude=('binproviders_supported',)), + 'loaded_binprovider': binprovider, + 'binproviders_supported': self.binproviders_supported, + }) except Exception as err: # print(err) inner_exc = err @@ -160,17 +154,22 @@ def load(self, cache=True) -> Self: if self.is_valid: return self - if not self.providers_supported: + if not self.binproviders_supported: return self - outer_exc = Exception(f'None of the configured providers [{", ".join(p.name for p in self.providers_supported)}] were able to load binary: {self.name}') + outer_exc = Exception(f'None of the configured providers [{", ".join(p.name for p in self.binproviders_supported)}] were able to load binary: {self.name}') inner_exc = Exception('No providers were available') - for provider in self.providers_supported: + for binprovider in self.binproviders_supported: try: - installed_bin = provider.load(self.name, cache=cache, overrides=self.provider_overrides.get(provider.name)) + installed_bin = binprovider.load(self.name, cache=cache, overrides=self.provider_overrides.get(binprovider.name)) if installed_bin: - # print('LOADED', provider, self.name, installed_bin) - return self.__class__.model_validate({**self.model_dump(), **installed_bin.model_dump(exclude=('providers_supported',))}) + # print('LOADED', binprovider, self.name, installed_bin) + return self.__class__.model_validate({ + **self.model_dump(), + **installed_bin.model_dump(exclude=('binproviders_supported',)), + 'loaded_binprovider': binprovider, + 'binproviders_supported': self.binproviders_supported, + }) except Exception as err: # print(err) inner_exc = err @@ -183,17 +182,22 @@ def load_or_install(self, cache=True) -> Self: if self.is_valid: return self - if not self.providers_supported: + if not self.binproviders_supported: return self - outer_exc = Exception(f'None of the configured providers [{", ".join(p.name for p in self.providers_supported)}] were able to find or install binary: {self.name}') + outer_exc = Exception(f'None of the configured providers [{", ".join(p.name for p in self.binproviders_supported)}] were able to find or install binary: {self.name}') inner_exc = Exception('No providers were available') - for provider in self.providers_supported: + for binprovider in self.binproviders_supported: try: - installed_bin = provider.load_or_install(self.name, overrides=self.provider_overrides.get(provider.name), cache=cache) + installed_bin = binprovider.load_or_install(self.name, overrides=self.provider_overrides.get(binprovider.name), cache=cache) if installed_bin: # print('LOADED_OR_INSTALLED', self.name, installed_bin) - return self.__class__.model_validate({**self.model_dump(), **installed_bin.model_dump(exclude=('providers_supported',))}) + return self.__class__.model_validate({ + **self.model_dump(), + **installed_bin.model_dump(exclude=('binproviders_supported',)), + 'loaded_binprovider': binprovider, + 'binproviders_supported': self.binproviders_supported, + }) except Exception as err: # print(err) inner_exc = err @@ -258,29 +262,29 @@ def get_ytdlp_version() -> str: class PythonBinary(Binary): name: BinName = 'python' - providers_supported: List[BinProvider] = [ + binproviders_supported: List[InstanceOf[BinProvider]] = [ EnvProvider( - packages_provider={'python': 'plugantic.binaries.SystemPythonHelpers.get_packages'}, - abspath_provider={'python': 'plugantic.binaries.SystemPythonHelpers.get_abspath'}, - version_provider={'python': 'plugantic.binaries.SystemPythonHelpers.get_version'}, + packages_handler={'python': 'plugantic.binaries.SystemPythonHelpers.get_packages'}, + abspath_handler={'python': 'plugantic.binaries.SystemPythonHelpers.get_abspath'}, + version_handler={'python': 'plugantic.binaries.SystemPythonHelpers.get_version'}, ), ] class SqliteBinary(Binary): name: BinName = 'sqlite' - providers_supported: List[BinProvider] = [ + binproviders_supported: List[InstanceOf[BinProvider]] = [ EnvProvider( - version_provider={'sqlite': 'plugantic.binaries.SqliteHelpers.get_version'}, - abspath_provider={'sqlite': 'plugantic.binaries.SqliteHelpers.get_abspath'}, + version_handler={'sqlite': 'plugantic.binaries.SqliteHelpers.get_version'}, + abspath_handler={'sqlite': 'plugantic.binaries.SqliteHelpers.get_abspath'}, ), ] class DjangoBinary(Binary): name: BinName = 'django' - providers_supported: List[BinProvider] = [ + binproviders_supported: List[InstanceOf[BinProvider]] = [ EnvProvider( - abspath_provider={'django': 'plugantic.binaries.DjangoHelpers.get_django_abspath'}, - version_provider={'django': 'plugantic.binaries.DjangoHelpers.get_django_version'}, + abspath_handler={'django': 'plugantic.binaries.DjangoHelpers.get_django_abspath'}, + version_handler={'django': 'plugantic.binaries.DjangoHelpers.get_django_version'}, ), ] @@ -290,17 +294,17 @@ class DjangoBinary(Binary): class YtdlpBinary(Binary): name: BinName = 'yt-dlp' - providers_supported: List[BinProvider] = [ + binproviders_supported: List[InstanceOf[BinProvider]] = [ # EnvProvider(), - PipProvider(version_provider={'yt-dlp': 'plugantic.binaries.YtdlpHelpers.get_ytdlp_version'}), - BrewProvider(packages_provider={'yt-dlp': 'plugantic.binaries.YtdlpHelpers.get_ytdlp_packages'}), - # AptProvider(packages_provider={'yt-dlp': lambda: ['yt-dlp', 'ffmpeg']}), + PipProvider(version_handler={'yt-dlp': 'plugantic.binaries.YtdlpHelpers.get_ytdlp_version'}), + BrewProvider(packages_handler={'yt-dlp': 'plugantic.binaries.YtdlpHelpers.get_ytdlp_packages'}), + # AptProvider(packages_handler={'yt-dlp': lambda: ['yt-dlp', 'ffmpeg']}), ] class WgetBinary(Binary): name: BinName = 'wget' - providers_supported: List[BinProvider] = [EnvProvider(), AptProvider()] + binproviders_supported: List[InstanceOf[BinProvider]] = [EnvProvider(), AptProvider()] # if __name__ == '__main__': diff --git a/pydantic_pkgr/binprovider.py b/pydantic_pkgr/binprovider.py index 0815a6f..ae3fe0b 100644 --- a/pydantic_pkgr/binprovider.py +++ b/pydantic_pkgr/binprovider.py @@ -12,16 +12,16 @@ from subprocess import run, PIPE, CompletedProcess from pydantic_core import core_schema, ValidationError -from pydantic import BaseModel, Field, TypeAdapter, AfterValidator, BeforeValidator, validate_call, GetCoreSchemaHandler, ConfigDict, computed_field, field_validator, model_validator +from pydantic import BaseModel, Field, TypeAdapter, AfterValidator, BeforeValidator, validate_call, GetCoreSchemaHandler, ConfigDict, computed_field, field_validator, model_validator, InstanceOf -def validate_bin_provider_name(name: str) -> str: +def validate_binprovider_name(name: str) -> str: assert 1 < len(name) < 16, 'BinProvider names must be between 1 and 16 characters long' assert name.replace('_', '').isalnum(), 'BinProvider names can only contain a-Z0-9 and underscores' assert name[0].isalpha(), 'BinProvider names must start with a letter' return name -BinProviderName = Annotated[str, AfterValidator(validate_bin_provider_name)] +BinProviderName = Annotated[str, AfterValidator(validate_binprovider_name)] # in practice this is essentially BinProviderName: Literal['env', 'pip', 'apt', 'brew', 'npm', 'vendor'] # but because users can create their own BinProviders we cant restrict it to a preset list of literal names @@ -173,19 +173,27 @@ class ShallowBinary(BaseModel): model_config = ConfigDict(extra='ignore', populate_by_name=True, validate_defaults=True) name: BinName = '' + description: str = '' - providers_supported: List['BinProvider'] = Field(default=[], alias='providers') + binproviders_supported: List[InstanceOf['BinProvider']] = Field(default=[], alias='binproviders') + provider_overrides: Dict[BinProviderName, 'ProviderLookupDict'] = Field(default={}, alias='overrides') - loaded_provider: BinProviderName = Field(default='env', alias='provider') + loaded_binprovider: InstanceOf['BinProvider'] = Field(alias='binprovider') loaded_abspath: HostBinPath = Field(alias='abspath') loaded_version: SemVer = Field(alias='version') + def __getattr__(self, item): """Allow accessing fields as attributes by both field name and alias name""" for field, meta in self.model_fields.items(): if meta.alias == item: return getattr(self, field) return super().__getattr__(item) + + @model_validator(mode='after') + def validate(self): + self.description = self.description or self.name + return self @computed_field # type: ignore[misc] # see mypy issue #1362 @property @@ -270,7 +278,7 @@ def is_valid_python_dotted_import(import_str: str) -> str: #ProviderHandlerStr = Annotated[str, AfterValidator(lambda s: s.startswith('self.'))] ProviderHandlerRef = LazyImportStr | ProviderHandler ProviderLookupDict = Dict[str, ProviderHandlerRef] -ProviderType = Literal['abspath', 'version', 'packages', 'install'] +HandlerType = Literal['abspath', 'version', 'packages', 'install'] # class Host(BaseModel): @@ -284,17 +292,16 @@ def is_valid_python_dotted_import(import_str: str) -> str: class BinProvider(BaseModel): - model_config = ConfigDict(extra='ignore', populate_by_name=True, validate_defaults=True) - + model_config = ConfigDict(extra='ignore', populate_by_name=True, validate_defaults=True, revalidate_instances='always') name: BinProviderName = '' PATH: PATHStr = Field(default='') # e.g. '/opt/homebrew/bin:/opt/archivebox/bin' INSTALLER_BIN: BinName = 'env' - abspath_provider: ProviderLookupDict = Field(default={'*': 'self.on_get_abspath'}, exclude=True) - version_provider: ProviderLookupDict = Field(default={'*': 'self.on_get_version'}, exclude=True) - packages_provider: ProviderLookupDict = Field(default={'*': 'self.on_get_packages'}, exclude=True) - install_provider: ProviderLookupDict = Field(default={'*': 'self.on_install'}, exclude=True) + abspath_handler: ProviderLookupDict = Field(default={'*': 'self.on_get_abspath'}, exclude=True) + version_handler: ProviderLookupDict = Field(default={'*': 'self.on_get_version'}, exclude=True) + packages_handler: ProviderLookupDict = Field(default={'*': 'self.on_get_packages'}, exclude=True) + install_handler: ProviderLookupDict = Field(default={'*': 'self.on_install'}, exclude=True) _abspath_cache: ClassVar = {} _version_cache: ClassVar = {} @@ -302,13 +309,19 @@ class BinProvider(BaseModel): def __getattr__(self, item): """Allow accessing fields as attributes by both field name and alias name""" + if item in ('__fields__', 'model_fields'): + raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{item}'") + for field, meta in self.model_fields.items(): if meta.alias == item: return getattr(self, field) return super().__getattr__(item) - def __str__(self) -> str: - return f'{self.INSTALLER_BIN.title()}Provider[{self.INSTALLER_BIN_ABSPATH or self.INSTALLER_BIN})]' + # def __str__(self) -> str: + # return f'{self.name.title()}Provider[{self.INSTALLER_BIN_ABSPATH or self.INSTALLER_BIN})]' + + # def __repr__(self) -> str: + # return f'{self.name.title()}Provider[{self.INSTALLER_BIN_ABSPATH or self.INSTALLER_BIN})]' @computed_field @property @@ -325,14 +338,14 @@ def INSTALLER_BIN_ABSPATH(self) -> HostBinPath | None: def is_valid(self) -> bool: return bool(self.INSTALLER_BIN_ABSPATH) - # def provider_version(self) -> SemVer | None: + # def installer_version(self) -> SemVer | None: # """Version of the actual underlying package manager (e.g. pip v20.4.1)""" # if self.name in ('env', 'vendor'): # return SemVer('0.0.0') # installer_binpath = Path(shutil.which(self.name)).resolve() # return bin_version(installer_binpath) - # def provider_host(self) -> Host: + # def installer_host(self) -> Host: # """Information about the host env, archictecture, and OS needed to select & build packages""" # p = platform.uname() # return Host( @@ -361,97 +374,97 @@ def load_PATH(cls, PATH: PATHStr) -> PATHStr: PATH = ':'.join([python_bin_dir, *PATH.split(':')]) return TypeAdapter(PATHStr).validate_python(PATH) - def get_default_providers(self): - return self.get_providers_for_bin('*') + def get_default_handlers(self): + return self.get_handlers_for_bin('*') - def resolve_provider_func(self, provider_func: ProviderHandlerRef | None) -> ProviderHandler | None: - if provider_func is None: + def resolve_handler_func(self, handler_func: ProviderHandlerRef | None) -> ProviderHandler | None: + if handler_func is None: return None - # if provider_func is already a callable, return it directly - if isinstance(provider_func, Callable): - return TypeAdapter(ProviderHandler).validate_python(provider_func) + # if handler_func is already a callable, return it directly + if isinstance(handler_func, Callable): + return TypeAdapter(ProviderHandler).validate_python(handler_func) - # if provider_func is a dotted path to a function on self, swap it for the actual function - if isinstance(provider_func, str) and provider_func.startswith('self.'): - provider_func = getattr(self, provider_func.split('self.', 1)[-1]) + # if handler_func is a dotted path to a function on self, swap it for the actual function + if isinstance(handler_func, str) and handler_func.startswith('self.'): + handler_func = getattr(self, handler_func.split('self.', 1)[-1]) - # if provider_func is a dot-formatted import string, import the function - if isinstance(provider_func, str): + # if handler_func is a dot-formatted import string, import the function + if isinstance(handler_func, str): try: from django.utils.module_loading import import_string except ImportError: from importlib import import_module import_string = import_module - package_name, module_name, classname, path = provider_func.split('.', 3) # -> abc, def, ghi.jkl + package_name, module_name, classname, path = handler_func.split('.', 3) # -> abc, def, ghi.jkl # get .ghi.jkl nested attr present on module abc.def imported_module = import_string(f'{package_name}.{module_name}.{classname}') - provider_func = operator.attrgetter(path)(imported_module) + handler_func = operator.attrgetter(path)(imported_module) # # abc.def.ghi.jkl -> 1, 2, 3 # for idx in range(1, len(path)): # parent_path = '.'.join(path[:-idx]) # abc.def.ghi # try: # parent_module = import_string(parent_path) - # provider_func = getattr(parent_module, path[-idx]) + # handler_func = getattr(parent_module, path[-idx]) # except AttributeError, ImportError: # continue - assert provider_func, ( - f'{self.__class__.__name__} provider func for {bin_name} was not a function or dotted-import path: {provider_func}') + assert handler_func, ( + f'{self.__class__.__name__} handler func for {bin_name} was not a function or dotted-import path: {handler_func}') - return TypeAdapter(ProviderHandler).validate_python(provider_func) + return TypeAdapter(ProviderHandler).validate_python(handler_func) @validate_call - def get_providers_for_bin(self, bin_name: str) -> ProviderLookupDict: - providers_for_bin = { - 'abspath': self.abspath_provider.get(bin_name), - 'version': self.version_provider.get(bin_name), - 'packages': self.packages_provider.get(bin_name), - 'install': self.install_provider.get(bin_name), + def get_handlers_for_bin(self, bin_name: str) -> ProviderLookupDict: + handlers_for_bin = { + 'abspath': self.abspath_handler.get(bin_name), + 'version': self.version_handler.get(bin_name), + 'packages': self.packages_handler.get(bin_name), + 'install': self.install_handler.get(bin_name), } - only_set_providers_for_bin = {k: v for k, v in providers_for_bin.items() if v is not None} + only_set_handlers_for_bin = {k: v for k, v in handlers_for_bin.items() if v is not None} - return only_set_providers_for_bin + return only_set_handlers_for_bin @validate_call - def get_provider_for_action(self, bin_name: BinName, provider_type: ProviderType, default_provider: Optional[ProviderHandlerRef]=None, overrides: Optional[ProviderLookupDict]=None) -> ProviderHandler: + def get_handler_for_action(self, bin_name: BinName, handler_type: HandlerType, default_handler: Optional[ProviderHandlerRef]=None, overrides: Optional[ProviderLookupDict]=None) -> ProviderHandler: """ - Get the provider func for a given key + Dict of provider callbacks + fallback default provider. - e.g. get_provider_for_action(bin_name='yt-dlp', 'install', default_provider=self.on_install, ...) -> Callable + Get the handler func for a given key + Dict of handler callbacks + fallback default handler. + e.g. get_handler_for_action(bin_name='yt-dlp', 'install', default_handler=self.on_install, ...) -> Callable """ - provider_func_ref = ( - (overrides or {}).get(provider_type) - or self.get_providers_for_bin(bin_name).get(provider_type) - or self.get_default_providers().get(provider_type) - or default_provider + handler_func_ref = ( + (overrides or {}).get(handler_type) + or self.get_handlers_for_bin(bin_name).get(handler_type) + or self.get_default_handlers().get(handler_type) + or default_handler ) - # print('getting provider for action', bin_name, provider_type, provider_func) + # print('getting handler for action', bin_name, handler_type, handler_func) - provider_func = self.resolve_provider_func(provider_func_ref) + handler_func = self.resolve_handler_func(handler_func_ref) - assert provider_func, f'No {self.name} provider func was found for {bin_name} in: {self.__class__.__name__}.' + assert handler_func, f'No {self.name} handler func was found for {bin_name} in: {self.__class__.__name__}.' - return provider_func + return handler_func @validate_call - def call_provider_for_action(self, bin_name: BinName, provider_type: ProviderType, default_provider: Optional[ProviderHandlerRef]=None, overrides: Optional[ProviderLookupDict]=None, **kwargs) -> Any: - provider_func: ProviderHandler = self.get_provider_for_action( + def call_handler_for_action(self, bin_name: BinName, handler_type: HandlerType, default_handler: Optional[ProviderHandlerRef]=None, overrides: Optional[ProviderLookupDict]=None, **kwargs) -> Any: + handler_func: ProviderHandler = self.get_handler_for_action( bin_name=bin_name, - provider_type=provider_type, - default_provider=default_provider, + handler_type=handler_type, + default_handler=default_handler, overrides=overrides, ) - if not func_takes_args_or_kwargs(provider_func): + if not func_takes_args_or_kwargs(handler_func): # if it's a pure argless lambdas, dont pass bin_path and other **kwargs - provider_func_without_args = cast(Callable[[], Any], provider_func) - return provider_func_without_args() + handler_func_without_args = cast(Callable[[], Any], handler_func) + return handler_func_without_args() - provider_func = cast(Callable[..., Any], provider_func) - return provider_func(bin_name, **kwargs) + handler_func = cast(Callable[..., Any], handler_func) + return handler_func(bin_name, **kwargs) def setup_PATH(self): for path in reversed(self.PATH.split(':')): @@ -504,10 +517,10 @@ def get_abspaths(self, bin_name: BinName) -> List[HostBinPath]: @validate_call def get_abspath(self, bin_name: BinName, overrides: Optional[ProviderLookupDict]=None) -> HostBinPath | None: self.setup_PATH() - abspath = self.call_provider_for_action( + abspath = self.call_handler_for_action( bin_name=bin_name, - provider_type='abspath', - default_provider=self.on_get_abspath, + handler_type='abspath', + default_handler=self.on_get_abspath, overrides=overrides, ) if not abspath: @@ -518,10 +531,10 @@ def get_abspath(self, bin_name: BinName, overrides: Optional[ProviderLookupDict] @validate_call def get_version(self, bin_name: BinName, abspath: Optional[HostBinPath]=None, overrides: Optional[ProviderLookupDict]=None) -> SemVer | None: - version = self.call_provider_for_action( + version = self.call_handler_for_action( bin_name=bin_name, - provider_type='version', - default_provider=self.on_get_version, + handler_type='version', + default_handler=self.on_get_version, overrides=overrides, abspath=abspath, ) @@ -533,10 +546,10 @@ def get_version(self, bin_name: BinName, abspath: Optional[HostBinPath]=None, ov @validate_call def get_packages(self, bin_name: BinName, overrides: Optional[ProviderLookupDict]=None) -> InstallArgs: - packages = self.call_provider_for_action( + packages = self.call_handler_for_action( bin_name=bin_name, - provider_type='packages', - default_provider=self.on_get_packages, + handler_type='packages', + default_handler=self.on_get_packages, overrides=overrides, ) if not packages: @@ -548,10 +561,10 @@ def get_packages(self, bin_name: BinName, overrides: Optional[ProviderLookupDict def install(self, bin_name: BinName, overrides: Optional[ProviderLookupDict]=None) -> ShallowBinary | None: packages = self.get_packages(bin_name, overrides=overrides) self.setup_PATH() - self.call_provider_for_action( + self.call_handler_for_action( bin_name=bin_name, - provider_type='install', - default_provider=self.on_install, + handler_type='install', + default_handler=self.on_install, overrides=overrides, packages=packages, ) @@ -564,10 +577,10 @@ def install(self, bin_name: BinName, overrides: Optional[ProviderLookupDict]=Non result = ShallowBinary( name=bin_name, - provider=self.name, + binprovider=self, abspath=installed_abspath, version=installed_version, - providers=[self], + binproviders=[self], ) self._install_cache[bin_name] = result return result @@ -595,10 +608,10 @@ def load(self, bin_name: BinName, overrides: Optional[ProviderLookupDict]=None, return ShallowBinary( name=bin_name, - provider=self.name, + binprovider=self, abspath=installed_abspath, version=installed_version, - providers=[self], + binproviders=[self], ) @validate_call @@ -608,7 +621,6 @@ def load_or_install(self, bin_name: BinName, overrides: Optional[ProviderLookupD installed = self.install(bin_name=bin_name, overrides=overrides) return installed - class PipProvider(BinProvider): name: BinProviderName = 'pip' INSTALLER_BIN: BinName = 'pip' @@ -639,6 +651,8 @@ def on_install(self, bin_name: str, packages: Optional[InstallArgs]=None, **cont print(proc.stdout.strip()) print(proc.stderr.strip()) raise Exception(f'{self.__class__.__name__}: install got returncode {proc.returncode} while installing {packages}: {packages}') + + class NpmProvider(BinProvider): name: BinProviderName = 'npm' @@ -691,8 +705,8 @@ class AptProvider(BinProvider): name: BinProviderName = 'apt' INSTALLER_BIN: BinName = 'apt-get' - packages_provider: ProviderLookupDict = { - **BinProvider.model_fields['packages_provider'].default, + packages_handler: ProviderLookupDict = { + **BinProvider.model_fields['packages_handler'].default, 'yt-dlp': lambda: ['yt-dlp', 'ffmpeg'], # always install ffmpeg when installing yt-dlp } @@ -791,14 +805,14 @@ class EnvProvider(BinProvider): INSTALLER_BIN: BinName = 'env' PATH: PATHStr = Field(default=DEFAULT_ENV_PATH) # add dir containing python to $PATH - abspath_provider: ProviderLookupDict = { - **BinProvider.__fields__['abspath_provider'].default, + abspath_handler: ProviderLookupDict = { + **BinProvider.model_fields['abspath_handler'].default, 'python': 'self.get_python_abspath', # 'sqlite': 'self.get_sqlite_abspath', # 'django': 'self.get_django_abspath', } - version_provider: ProviderLookupDict = { - **BinProvider.__fields__['version_provider'].default, + version_handler: ProviderLookupDict = { + **BinProvider.model_fields['version_handler'].default, 'python': 'self.get_python_version', # 'sqlite': 'self.get_sqlite_version', # 'django': 'self.get_django_version', @@ -833,5 +847,5 @@ def get_python_version(): # return '{}.{}.{} {} ({})'.format(*django.VERSION) def on_install(self, bin_name: BinName, packages: Optional[InstallArgs]=None, **context): - """The env provider is ready-only and does not install any packages, so this is a no-op""" + """The env BinProvider is ready-only and does not install any packages, so this is a no-op""" pass diff --git a/pyproject.toml b/pyproject.toml index 8c15b16..89bc441 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pydantic-pkgr" -version = "0.2.0" +version = "0.2.1" description = "System package manager APIs in strongly typed Python" authors = [ {name = "Nick Sweeting", email = "pydantic-pkgr-pyproject-toml@sweeting.me"}, diff --git a/tests.py b/tests.py index c5f145e..1b6e483 100644 --- a/tests.py +++ b/tests.py @@ -60,11 +60,11 @@ def test_python_env(self): def test_bash_env(self): - provider = EnvProvider() + envprovider = EnvProvider() SYS_BASH_VERSION = subprocess.check_output('bash --version', shell=True, text=True).split('\n')[0] - bash_bin = provider.load_or_install('bash') + bash_bin = envprovider.load_or_install('bash') self.assertEqual(bash_bin.loaded_version, SemVer(SYS_BASH_VERSION)) self.assertGreater(bash_bin.loaded_version, SemVer('3.0.0')) self.assertEqual(bash_bin.loaded_abspath, Path(shutil.which('bash'))) @@ -85,16 +85,16 @@ class TestRecord: class CustomProvider(BinProvider): name: str = 'CustomProvider' - abspath_provider: ProviderLookupDict = { + abspath_handler: ProviderLookupDict = { '*': 'self.on_abspath_custom' } - version_provider: ProviderLookupDict = { + version_handler: ProviderLookupDict = { '*': 'self.on_version_custom' } - packages_provider: ProviderLookupDict = { + packages_handler: ProviderLookupDict = { '*': 'self.on_packages_custom' } - install_provider: ProviderLookupDict = { + install_handler: ProviderLookupDict = { '*': 'does.not.exist' } @@ -117,7 +117,7 @@ def on_install(self, bin_name: str, **context): def on_install_somebin(self, bin_name: str, **context): TestRecord.called_install_custom = True - provider = CustomProvider(install_provider={'somebin': 'self.on_install_somebin'}) + provider = CustomProvider(install_handler={'somebin': 'self.on_install_somebin'}) self.assertFalse(TestRecord.called_abspath_custom) self.assertFalse(TestRecord.called_version_custom) @@ -148,18 +148,19 @@ def on_install_somebin(self, bin_name: str, **context): class TestBinary(unittest.TestCase): def test_python_bin(self): - provider = EnvProvider() + envprovider = EnvProvider() - python_bin = Binary(name='python', providers=[provider]) + python_bin = Binary(name='python', binproviders=[envprovider]) - self.assertIsNone(python_bin.loaded_provider) + self.assertIsNone(python_bin.loaded_binprovider) self.assertIsNone(python_bin.loaded_abspath) self.assertIsNone(python_bin.loaded_version) python_bin = python_bin.load() - shallow_bin = provider.load_or_install('python') - self.assertEqual(python_bin.loaded_provider, shallow_bin.loaded_provider) + shallow_bin = envprovider.load_or_install('python') + assert shallow_bin and python_bin.loaded_binprovider + self.assertEqual(python_bin.loaded_binprovider, shallow_bin.loaded_binprovider) self.assertEqual(python_bin.loaded_abspath, shallow_bin.loaded_abspath) self.assertEqual(python_bin.loaded_version, shallow_bin.loaded_version) @@ -177,8 +178,7 @@ def flatten(xss): class InstallTest(unittest.TestCase): - def install_with_provider(self, provider, binary): - + def install_with_binprovider(self, provider, binary): binary_bin = binary.load_or_install() provider_bin = provider.load_or_install(bin_name=binary.name) @@ -186,7 +186,7 @@ def install_with_provider(self, provider, binary): # print('\n'.join(f'{provider}={path}' for provider, path in binary.loaded_abspaths.items()), '\n') # print() - self.assertEqual(binary_bin.loaded_provider, provider_bin.loaded_provider) + self.assertEqual(binary_bin.loaded_binprovider, provider_bin.loaded_binprovider) self.assertEqual(binary_bin.loaded_abspath, provider_bin.loaded_abspath) self.assertEqual(binary_bin.loaded_version, provider_bin.loaded_version) @@ -214,20 +214,20 @@ def install_with_provider(self, provider, binary): def test_env_provider(self): provider = EnvProvider() - binary = Binary(name='wget', providers=[provider]).load() - self.install_with_provider(provider, binary) + binary = Binary(name='wget', binproviders=[provider]).load() + self.install_with_binprovider(provider, binary) def test_pip_provider(self): - provider = PipProvider() + pipprovider = PipProvider() # print(provider.PATH) - binary = Binary(name='wget', providers=[provider]) - self.install_with_provider(provider, binary) + binary = Binary(name='wget', binproviders=[pipprovider]) + self.install_with_binprovider(pipprovider, binary) def test_npm_provider(self): - provider = NpmProvider() + npmprovider = NpmProvider() # print(provider.PATH) - binary = Binary(name='wget', providers=[provider]) - self.install_with_provider(provider, binary) + binary = Binary(name='wget', binproviders=[npmprovider]) + self.install_with_binprovider(npmprovider, binary) def test_brew_provider(self): # print(provider.PATH) @@ -250,8 +250,8 @@ def test_brew_provider(self): exception = None result = None try: - binary = Binary(name='wget', providers=[provider]) - result = self.install_with_provider(provider, binary) + binary = Binary(name='wget', binproviders=[provider]) + result = self.install_with_binprovider(provider, binary) except Exception as err: exception = err @@ -287,8 +287,8 @@ def test_apt_provider(self): self.assertFalse(provider.PATH) try: # print(provider.PATH) - binary = Binary(name='wget', providers=[provider]) - result = self.install_with_provider(provider, binary) + binary = Binary(name='wget', binproviders=[provider]) + result = self.install_with_binprovider(provider, binary) except Exception as err: exception = err