Source code for pspman.classes

#!/usr/bin/env python3
# -*- coding:utf-8; mode:python -*-
#
# Copyright 2020 Pradyumna Paranjape
# This file is part of pspman.
#
# pspman is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# pspman is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with pspman.  If not, see <https://www.gnu.org/licenses/>.
#
'''
object classes

'''


import typing
import re
from datetime import timezone, datetime
from pathlib import Path
import json
import yaml
from psprint import print
from .config import MetaConfig
from .tag import ACTION_TAG
from .errors import GitURLError


[docs]class InstallEnv(): ''' Installation variables Attributes: call_function: sub-function called {version,info,meta,unlock} clone_dir: base directory to clone src prefix: `prefix` for installation risk: risk root pull: only pull, don't try to install stale: do not update, only delete/install new verbose: print semi-verbose output install: install (clone) from remote urls delete: delete projects Args: config: MetaConfig to determine default values **kwargs: hard set `attributes` ''' def __init__(self, config: MetaConfig, **kwargs): self.call_function: typing.Optional[str] = kwargs.get('call_function') self._clone_dir: typing.Optional[Path] = kwargs.get('clone_dir') self._prefix: str = kwargs.get('prefix', config.data_dir) self.risk: bool = kwargs.get('risk', False) self.pull: bool = kwargs.get('pull', False) self.stale: bool = kwargs.get('stale', False) self.verbose: bool = kwargs.get('verbose', False) self.reset: typing.List[str] = kwargs.get('reset', []) self.install: typing.List[str] = kwargs.get('install', []) self.delete: typing.List[str] = kwargs.get('delete', []) @property def prefix(self) -> Path: return Path(self._prefix) @prefix.setter def prefix(self, value): self._prefix = str(value) @property def clone_dir(self) -> Path: if self._clone_dir is None: return self.prefix.joinpath('src') return Path(self._clone_dir) @clone_dir.setter def clone_dir(self, value): self._clone_dir = value @clone_dir.deleter def clone_dir(self): self._clone_dir = None def __repr__(self) -> str: return f''' Clone Directory: {self.clone_dir} Prefix: {self.prefix} Called sub-function: {self.call_function} Clean-Resets Requested: {len(self.reset)} Deletions Requested: {len(self.delete)} Additions requested: {len(self.install)} Optional Flags: Risk Root: {self.risk} Only Pull: {self.pull} Don't Update: {self.stale} Verbose Debugging: {self.verbose} '''
[docs] def update(self, options: typing.Dict[str, object]) -> 'InstallEnv': ''' Update attributes with those from options Args: options: hard-set config ''' for key, value in options.items(): if hasattr(self, key): setattr(self, key, value) return self
[docs]class GitProject(): ''' Git project object. Args: **kwargs: * args corresponding to Attributes: hard set * data: Dict[str, Any]: load data * rest are ignored Attributes: url: url for git remote name: name of project folder tag: action tagged to project branch: git branch to be cloned sh_env: environ modifications before installation inst_argv: arguments suffixed to optional args before positional args pull: only pull this project, don't run install scripts last_updated: last updated on datetime ''' def __init__(self, **kwargs) -> None: self.url: str = kwargs.get('url', None) self.tag = int(kwargs.get('tag', 0)) self.branch: typing.Optional[str] = kwargs.get('branch') self.last_updated: typing.Optional[float] = kwargs.get('last_updated') self.inst_argv: typing.List[str] = kwargs.get('inst_argv', []) self.sh_env: typing.Dict[str, str] = kwargs.get('sh_env', {}) self.pull: bool = kwargs.get('pull', False) # Infer from parent.__dict__ if kwargs.get('data') is not None: self.merge(kwargs['data']) del kwargs['data'] self.name: str = kwargs.get('name', self.update_name())
[docs] def update_name(self) -> str: ''' Update project name. Call this method after updating ``url`` Returns: The updated name ''' if hasattr(self, 'name'): return self.name if self.url is None: raise GitURLError self.name = Path(self.url.replace(':', '/').split('/')[-1]).stem return self.name
[docs] def merge(self, data: typing.Dict[str, object]) -> None: ''' Update values that are ``None`` type using ``data``. Doesn't change set values Args: data: source for update ''' for key, val in data.items(): self.__dict__[key] = self.__dict__.get(key) or val
[docs] def mark_update_time(self): ''' Mark that the project was updated ''' self.last_updated = datetime.now().timestamp()
def __repr__(self) -> str: ''' representation of GitProject object ''' updated = datetime.fromtimestamp(self.last_updated)\ if self.last_updated else None return f''' *** {self.name} *** Source: {self.url} Branch: {self.branch} Last Updated: {updated} Only Pull?: {self.pull} Base tag: {hex(self.tag)} Installation arguments: {self.inst_argv} Altered shell environment variables: {self.sh_env} ''' def __str__(self) -> str: ''' Print object name ''' return str(self.name)
[docs]class GitProjEncoder(json.JSONEncoder): ''' Encode Class object's __dict__ '''
[docs] def default(self, o: GitProject) -> dict: return o.__dict__