Skip to content

Latest commit

 

History

History
3350 lines (2400 loc) · 89.8 KB

Reference.md

File metadata and controls

3350 lines (2400 loc) · 89.8 KB

Table of Contents

pylightnix.types

All main types which we use in Pylightnix are defined here.

Path Objects

Path is an alias for string. It is used in pylightnix to tell the typechecker that a given string contains a filesystem path.

SPath Objects

SPath is an alias for string. It is used in pylightnix to tell the typechecker that a given string contains a path to storage.

StorageSettings

StorageSettings = NamedTuple('StorageSettings',[('root',Optional[Path]),
                                              ...

Stoarge settings contains a path for the main stoarge and a path for temporary directories. These paths need to be on the same device in order to atomic rename work.

Hash Objects

Hash is an alias for string. It is used in pylightnix to tell the typechecker that a given string contains sha256 hash digest.

HashPart Objects

HashPart is an alias for string. It is used in pylightnix to tell the typechecker that a given string contains first 32 characters of sha256 hash digest.

DRef Objects

DRef stands for derivation reference. It is a string identifier of a filesystem part of Derivation object.

The format of derivation reference is <HashPart>-<Name>, where:

  • <HashPart> contains first 32 characters of derivation RConfig's sha256 hash digest.
  • <Name> object contains the name of derivation.

Derivation reference 'points to' derivation object in pylightnix filesystem storage. For a valid DRef, $PYLIGHTNIX_STORE/<HashPart>-<Name>/ does exist and is a directory which contains config.json file.

Derivation references are results of instantiation.

Derivation reference may be converted into a realization reference by either dereferencing (that is by querying for existing realizations) or by realizing it from scratch.

RRef Objects

RRef stands for Realization Reference. RRefs identify collections of artifacts of a Stage. Stages with non-determenistic realizers may have several competing realization instances. Every such instance is identified by a unique RRef.

The format of realization reference is <HashPart0>-<HashPart1>-<Name>, where:

  • <HashPart0> is calculated over realization's Context and build artifacts.
  • <HashPart1>-<Name> forms valid DRef which this realizaion was realized from.

Realization reference is created during the realization process .

Valid realization references may be dereferenced down to system paths of build artifacts by calling rref2path or by using lenses.

Autostage decorator unwraps RRefs of parent stages into Attrs objects.

Name Objects

Name is an alias for string. It is used in pylightnix to tell the typechecker that a given string contains name of a pylightnix storage object.

Names are restircted to contain charaters matching PYLIGHTNIX_NAMEPAT.

See also mkname

RefPath

RefPath = List[Union[DRef,str]]

RefPath is an alias for Python list (of strings). The first item of RefPath is a derivation reference. Other elements are to represent parts of file path. RefPath is designed to be used in a stage config where they typically refer to artifacts of already existing dependencies. To refer to future artifacts of the derivation being configured, use PromisePaths.

To convert RefPath into a system path, one generally have to perform the following basic actions:

  1. Dereference it's first item to obtain the realization. See store_deref or build_deref.
  2. Convert the realization reference into system path with rref2path
  3. Join the system path with [1:] part of RefPath to get the real filename.

The algorithm described above is implemented as build_path helper function.

PylightnixException Objects

Base class of Pylightnix exceptions

PromiseException Objects

def __init__(self, dref: DRef, failed: List[Tuple[Path,RefPath]])

PromiseException.__init__()

def __init__(self, dref: DRef, failed: List[Tuple[Path,RefPath]])

_REF

_REF = TypeVar('_REF')

Output Objects

def __init__(self, val: Iterable[_REF])

Output is a base class for 'organized collections of realizations', either in form of temporary Paths or RRefs.

TODO: Rename into something which has a meaning of PromisedOuput

Output.__init__()

def __init__(self, val: Iterable[_REF])

Context

Context = Dict[DRef,List[RRef]]

InstantiateArg

InstantiateArg = Dict[str,Any]

Type of user-defined arguments to pass to the Config

RealizeArg

RealizeArg = Dict[str,Any]

Type of user-defined arguments to pass to the Realizer

Matcher

Matcher = Callable[[Optional[StorageSettings],List[RRef]],
                   Optional[List[RRef]]]

Matchers are user-defined Python functions with the fixed signature. They serve two purposes:

  1. Decide whether to launch a new realization or re-use the results of realizations completed earlier.
  2. Filter a subset of a realizations to depend on out of the set of available realizations.

Matchers answer 'yes' to the first question by returning None. Non-none value specifies the matched set of realizations.

The Matcher's invariants are:

  • Matcher outputs should only depend on the immutable realizations passed to them as inputs. Matchers should avoid having side-effects.
  • Matchers must be satisfiable. If the matcher returns None, the core re-runs runs the realization and calls the matcher once again. Returning None again would be an error.

Pylightnix includes a set of built-in matchers:

MatcherO

MatcherO = Callable[[Optional[StorageSettings],Output[RRef]],
                    Optional[Output[RRef]]]

Realizer

Realizer = Callable[[Optional[StorageSettings],DRef,Context,RealizeArg],List[Path]]

Realizers are user-defined Python functions. Realizers typically implement application-specific algorithms which take some configuration parameters and produce some artifacts.

Realizer accepts the following arguments:

  • Path to a global Pylightnix storage
  • A Derivation reference being built
  • A Context encoding the results of dependency resolution.
  • Set of additional user-defined arguments

Context is the key to accessing the dependency artifacts.

Derivation reference is required to access configuration parameters of the algorithm.

Realizers must return one or many folders of realization artifacts (files and folders containing application-specific data). Every folder is treated as an alternative realization. Matcher is later used to pick the subset of realizations which matches some application-specific criteria. This subset will eventually appear as the Contexts of downstream realizaions.

Pylightnix stages may use the simplified realizer API provided by the Build helper class.

Example:

def mystage(r:Registry)->DRef:
def _realize(dref:DRef, context:Context)->List[Path]:
b=mkbuild(dref, context, buildtime=buildtime)
with open(join(build_outpath(b),'artifact'),'w') as f:
f.write('chickenpoop\n')
return [build_outpath(b)]
...
return mkdrv(r, ...,  _realize)

RealizerO

RealizerO = Callable[[Optional[StorageSettings],DRef,Context,RealizeArg],Output[Path]]

Derivation

Derivation = NamedTuple('Derivation', [('dref',DRef),
                                       ('matcher',Matcher), ...

Derivation is a core Pylightnix entity. It holds the information required to produce artifacts of individual Stage.

Fields include:

The actual configuration is stored in the Pylightnix filesystem storage. Derivation holds the DRef access key.

Derivations normally appear as a result of mkdrv calls.

Closure

Closure = NamedTuple('Closure', [('result',Any),
                                 ('targets',List[DRef]),
     ...

Closure describes the build plan for one or many Derivations.

Closures are typically obtained as a result of the instantiate and are consumed by the call to realize or it's analogs.

Config Objects

def __init__(self, d: dict)

Config is a JSON-serializable dict-like object containing user-defined attributes. Together with Realizers and Matchers, configs describe Stage objects.

Configs carry Python dictionaries that should contain JSON-serializable types. Strings, bools, ints, floats, lists or other dicts are fine, but no bytes, numpy.float32 or lambdas are allowed. Tuples are also forbidden because they are not preserved (decoded into lists). Special emphasis is placed on DRef support which link dependent stages together.

Config of a derivation can't include the Derivation reference to itself, because it contains the config hash as its part.

Some field names of a config have a special meaning for Pylightnix:

  • String name field will be used as a part of references to a derivation associated with this config.
  • RefPaths represent paths to artifacts within the stage artifact folders.
  • SelfRef paths represent output paths to be produced during the stage's realization.
  • DRef represent stage dependencies. Pylightnix collects derivation references and plan the realization order based on them.

Storing an RRef in the config leads to a warning. Pylightnix does not necessarily knows how to produce the exact reference, so the end result may not match the expectations.

Configs are normally created from Python dicts by the mkconfig function.

Example:

def mystage(r:Registry)->Dref:
  def _config()->dict:
    name = 'mystage'
    nepoches = 4
    learning_rate = 1e-5
    hidden_size = 128
    return locals()
  return mkdrv(mkconfig(_config()),...)

Config.__init__()

def __init__(self, d: dict)

Config.__repr__()

def __repr__(self) -> str

RConfig Objects

RConfig is a Config where all Self-referenes are resolved. RConfig stands for 'Resolved Config'.

ConfigAttrs Objects

def __init__(self, d: dict)

ConfigAttrs is a helper object allowing to access RConfig fields as Python object attributes.

DEPRECATED in favour of Lenses.

ConfigAttrs.__init__()

def __init__(self, d: dict)

BuildArgs

BuildArgs = NamedTuple('BuildArgs', [('S',Optional[StorageSettings]),
                                     ('dre ...

Build Objects

def __init__(self, ba: BuildArgs) -> None

Build objects track the process of stage's realization. Build allows users to define Realizers with only a simple one-argument signature. The build_wrapper function converts simplified Build-realizers into the regular ones.

Typical Build operations include:

  • build_config - Obtain the RConfig object of the current stage
  • build_cattrs - Obtain the ConfigAttrs helper
  • build_path - Convert a RefPath or a self-ref path into a system file path
  • build_setoutgroups - Initialize and return groups of output folders
  • build_deref - Convert a dependency DRef into a realization reference.

Lenses accept Build objects as a configuration source for derivations being realized.

Build class may be subclassed by applications in order to define application-specific build-state. Underscoped build_wrapper_ accepts additional callback parameter which informs the core what subclass to create. Note that derived classes should have the same constructor def __init__(self, ba:BuildArgs)->None.

Example:

class TensorFlowModel(Build):
  model:tf.keras.Model

def train(r:TensorFlowModel)->None:
  o = build_outpath(r)
  r.model = create_model(...)
  ...

def mymodel(r:Registry)->DRef:
  return mkdrv(r, ..., build_wrapper_(TensorFlowModel, train))

Build.__init__()

def __init__(self, ba: BuildArgs) -> None

Registry Objects

def __init__(self, S: Optional[StorageSettings] = None)

The derivation registry is a mutable storage object where Pylightnix stores derivations before combining them into a Closure.

Registry doesn't requre any special operations besides creating and passing around. By convention, Registry objects are first arguments of user-defined stage functions and the mkdrv API function of Pylightnix.

Registry.__init__()

def __init__(self, S: Optional[StorageSettings] = None)

DRefLike

DRefLike = TypeVar('DRefLike',bound=DRef)

StageResult

StageResult = Union[DRef,List[DRef],Dict[Any,DRef],Tuple[DRef,...]]

Stage

Stage = Callable[...,StageResult]

Pylightnix Stage is a signature of a Python function which creates a Derivations object and registers it in a Registry.

Stages are the top-level building blocks of Pylightnix.

Normally, stage functions create derivations by calling mkdrv one or many times and returning a StageResult value which is a name for a container holding one or many DRefs

A stage A may be set to depend on another stage B by including the DRefs of B into the configuration part of the A's derivation.

In order to run the registered stage(s), user has to pass the collectoin of DRefs to be realized to instantiate followed by the call to realize functions.

autostage decorator turns a Python function into a Pylightnix stage.

Pylightnix defines a number of built-in stages:

A note on typing: Real stages often accept additional custom arguments which AFAIK couldn't be handled by the simple MyPy. In a somewhat extended MyPy the Stage definition would look like:

Stage = Callable[[Registry,VarArg(Any),KwArg(Any)],StageResult]

Stage's return value is a derivation reference which could be either used in other stages, or instantiated into the stage realization plan.

pylightnix.core

Core Pylightnix definitions

PYLIGHTNIX_STORE_VERSION

PYLIGHTNIX_STORE_VERSION = 0

Do not change! Tracks the version of pylightnix storage

PYLIGHTNIX_NAMEPAT

PYLIGHTNIX_NAMEPAT = "[a-zA-Z0-9_-]"

Set the regular expression pattern for valid name characters.

PYLIGHTNIX_RESERVED

PYLIGHTNIX_RESERVED = ['context.json','group.json']

Reserved file names are treated specially be the core. Users should not normally create or alter files with these names.

logger

logger = getLogger(__name__)

info

info = logger.info

warning

warning = logger.warning

TL

TL = threading_local()

Thread-local storage for current_registry to store its state.

tlregistry()

def tlregistry(M: Optional[Registry]) -> Optional[Registry]

Return the currently active Registry

tlstorage()

def tlstorage(S: Optional[StorageSettings]) -> Optional[StorageSettings]

Return the currently active StorageSettings

storagename()

def storagename()

Return the name of Pylightnix storage filder.

fsroot()

def fsroot(S: Optional[StorageSettings] = None) -> Path

fsroot contains the path to the root of pylightnix shared data folder. Default is ~/_pylightnix or /var/run/_pylightnix if no $HOME is available. Setting PYLIGHTNIX_ROOT environment variable overwrites the defaults.

fstmpdir()

def fstmpdir(S: Optional[StorageSettings] = None) -> Path

Return the location of current Pylightnix temporary folder, defaulting to the path set by PYLIGHTNIX_TMP environment variable.

fsstorage()

def fsstorage(S: Optional[StorageSettings] = None) -> Path

Return the location of current Pylightnix storage folder, defaulting to the path set by PYLIGHTNIX_STORAGE environment variable.

assert_valid_storage()

def assert_valid_storage(S: Optional[StorageSettings] = None) -> None

mkSS()

def mkSS(root: str, stordir: Optional[str] = None, tmpdir: Optional[str] = None) -> StorageSettings

Constructor for StorageSettings

setstorage()

def setstorage(S: Optional[StorageSettings]) -> Optional[StorageSettings]

setregistry()

def setregistry(r: Optional[Registry]) -> Optional[Registry]

fsinit()

def fsinit(ss: Optional[Union[str,StorageSettings]] = None, check_not_exist: bool = False, remove_existing: bool = False, use_as_default: bool = False) -> None

Imperatively create the filesystem storage and temp direcory if they don't exist. Default locations may be altered by PYLIGHTNIX_STORAGE and PYLIGHTNIX_TMP env variables.

reserved()

def reserved(folder: Path, name: str) -> Path

trimhash()

def trimhash(h: Hash) -> HashPart

Trim a hash to get HashPart objects which are used in referencing

mkdref()

def mkdref(dhash: HashPart, refname: Name) -> DRef

rref2dref()

def rref2dref(rref: RRef) -> DRef

undref()

def undref(r: DRef) -> Tuple[HashPart, Name]

mkrref()

def mkrref(rhash: HashPart, dhash: HashPart, refname: Name) -> RRef

unrref()

def unrref(r: RRef) -> Tuple[HashPart, HashPart, Name]

mkname()

def mkname(s: str) -> Name

path2dref()

def path2dref(p: Path) -> Optional[DRef]

Takes either a system path of some realization in the Pylightnix storage or a symlink pointing to such path. Return a DRef which corresponds to this path.

Note: path2dref operates on p symbolically. It doesn't actually check the presence of such an object in storage

path2rref()

def path2rref(p: Path) -> Optional[RRef]

Takes either a system path of some realization in the Pylightnix storage or a symlink pointing to such path. Return RRef which corresponds to this path.

Note: path2rref operates on p symbolically. It doesn't actually check the presence of such an object in storage

mkconfig()

def mkconfig(d: dict) -> Config

Create a Config object out of config dictionary. Asserts if the dictionary is not JSON-compatible. As a handy hack, filter out r:Registry variable which likely is an utility Registry object.

cfgdict()

def cfgdict(cp: Config) -> dict

cfgcattrs()

def cfgcattrs(c: RConfig) -> Any

cfgserialize()

def cfgserialize(c: Config) -> str

cfghash()

def cfghash(c: Config) -> Hash

cfgname()

def cfgname(c: Config) -> Name

Return a name field of a config c, defaulting to string "unnmaed".

cfgdeps()

def cfgdeps(c: Config) -> Set[DRef]

mkrefpath()

def mkrefpath(r: DRef, items: List[str] = []) -> RefPath

Construct a RefPath out of a reference ref and a path within the stage's realization

resolve()

def resolve(c: Config, r: DRef) -> RConfig

Replace all Promise tags with DRef r. In particular, all PromisePaths are converted into RefPaths.

dref2path()

def dref2path(r: DRef, S=None) -> Path

rref2path()

def rref2path(r: RRef, S=None) -> Path

rrefpath2path()

def rrefpath2path(r: RRef, refpath: RefPath, S=None) -> Path

drefcfgpath()

def drefcfgpath(r: DRef, S=None) -> Path

rrefctx()

def rrefctx(r: RRef, S=None) -> Context

Return the realization context.

drefcfg_()

def drefcfg_(dref: DRef, S=None) -> Config

Return dref configuration, selfrefs are not resolved

drefcfg()

def drefcfg(dref: DRef, S=None) -> RConfig

Return dref configuration, selfrefs are resolved

drefattrs()

def drefattrs(r: DRef, S=None) -> Any

Read the ConfigAttrs of the storage node r. Note, that it is a kind of 'syntactic sugar' for drefcfg. Both functions do the same thing.

rrefattrs()

def rrefattrs(r: RRef, S=None) -> Any

Read the ConfigAttrs of the storage node r. Note, that it is a kind of 'syntactic sugar' for drefcfg. Both functions do the same thing.

drefdeps1()

def drefdeps1(drefs: Iterable[DRef], S=None) -> Set[DRef]

Return a set of reference's immediate dependencies, not including drefs themselves.

rrefdeps1()

def rrefdeps1(rrefs: Iterable[RRef], S=None) -> Set[RRef]

Return a set of reference's immediate dependencies, not including rrefs themselves.

drefdeps()

def drefdeps(drefs: Iterable[DRef], S=None) -> Set[DRef]

Return the complete set of drefs's dependencies, not including drefs themselves.

rrefdeps()

def rrefdeps(rrefs: Iterable[RRef], S=None) -> Set[RRef]

Return the complete set of rrefs's dependencies, not including rrefs themselves.

TODO: Validate the property that the resulting set IS the minimal complete set of RRef dependencies. Now it looks so only by creation (see realizeSeq, line mark I)

alldrefs()

def alldrefs(S=None) -> Iterable[DRef]

Iterates over all derivations of the storage located at S (PYLIGHTNIX_STORE env is used by default)

allrrefs()

def allrrefs(S=None) -> Iterable[RRef]

Iterates over all realization references in S (PYLIGHTNIX_STORE env is used by default)

rootdrefs()

def rootdrefs(S: Optional[StorageSettings] = None) -> Set[DRef]

Return root DRefs of the storage S as a set

rootrrefs()

def rootrrefs(S: Optional[StorageSettings] = None) -> Set[RRef]

Return root RRefs of the storage S as a set

rrefdata()

def rrefdata(rref: RRef, S=None) -> Iterable[Path]

Iterate over top-level artifacts paths, ignoring reserved files.

drefrrefs()

def drefrrefs(dref: DRef, S=None) -> Set[RRef]

Iterate over all realizations of a derivation dref. The sort order is unspecified. Matchers are not taken into account.

drefrrefsC()

def drefrrefsC(dref: DRef, context: Context, S=None) -> Iterable[RRef]

Iterate over realizations of a derivation dref that match the specified context. Sorting order is unspecified.

store_gc()

def store_gc(keep_drefs: List[DRef], keep_rrefs: List[RRef], S: Optional[StorageSettings] = None) -> Tuple[Set[DRef],Set[RRef]]

Take roots which are in use and should not be removed. Return roots which are not used and may be removed. Actual removing is to be done by the user.

Default location of S may be changed.

See also rmref

mkdrv_()

def mkdrv_(c: Config, S=None) -> DRef

See mkdrv

mkrealization()

def mkrealization(dref: DRef, l: Context, o: Path, S=None) -> RRef

Inserts the newly-obtaind Stage artifacts into the Storage, return the realization reference. Not intended to be called by user.

Parameters:

  • dref:DRef: Derivation reference to create the realization of.
  • l:Context: Context which stores dependency information.
  • o:Path: Path to temporal (build) folder which contains artifacts, prepared by the Realizer.
  • leader: Tag name and Group identifier of the Group leader. By default, we use name out and derivation's own rref.

mkcontext()

def mkcontext() -> Context

context_eq()

def context_eq(a: Context, b: Context) -> bool

context_add()

def context_add(ctx: Context, dref: DRef, rrefs: List[RRef]) -> Context

Add a pair (dref,rrefs) into a context ctx. rrefs are supposed to form (a subset of) the realizations of dref. Return a new context.

context_deref()

def context_deref(context: Context, dref: DRef) -> List[RRef]

TODO: Should it return Output (aka UniformList) rather than Python list?

context_derefpath()

def context_derefpath(context: Context, refpath: RefPath, S=None) -> List[Path]

context_serialize()

def context_serialize(c: Context) -> str

output_validate()

def output_validate(dref: DRef, o: Output[Path], S=None) -> List[Path]

output_realizer()

def output_realizer(f: RealizerO) -> Realizer

output_matcher()

def output_matcher(r: MatcherO) -> Matcher

mkdrv()

def mkdrv(config: Config, matcher: Matcher, realizer: Realizer, r: Optional[Registry] = None) -> DRef

Construct a Derivation object out of Config, Matcher and Realizer. Register the derivation in the dependency-resolution Registry. Return Derivation references of the newly-obtained derivation.

Arguments:

  • r:Registry: A Registry to update with a new derivation

Example:

def somestage(r:Registry)->DRef:
  def _realizer(b:Build):
    with open(join(build_outpath(b),'artifact'),'w') as f:
      f.write(...)
  return mkdrv(r,mkconfig({'name':'mystage'}), match_only(), build_wrapper(_realizer))

rref:RRef=realize1(instantiate(somestage))

current_registry()

@contextmanager
def current_registry(r: Registry) -> Iterator[Registry]

Sets the default global Registry for the inner scoped code. Internally calls current_storage(r.S)

current_storage()

@contextmanager
def current_storage(S: Optional[StorageSettings]) -> Iterator[Optional[StorageSettings]]

Sets the global default StorageSettings for the inner scoped code

mkclosure()

def mkclosure(result: Any, r: Registry) -> Closure

_A

_A = TypeVar('_A')

instantiate()

def instantiate(stage: Union[_A,Callable[...,Any]], args: Any, *,, ,, =, ,, =, ,, kwargs: Any) -> Tuple[_A,Closure]

Scans a Python DRef container (list, dict alike) or evaluates the Stage function by calling it.

Returns the Closure formed out of nested Derivations. The closure returned is ready to be realized.

Arguments:

  • stage: Stage function to call or DRef container to scan.
  • r:Optional[Registry]=None: Registry to register derivations in.
  • S:Optional[StorageSettings]=None: StorageSettings specifies the locations of the on-disk derivation data folders. Should match r.S if r is passed.

Returns:

  • DRef container holding references to the newly registered derivations.

RealizeSeqGen

RealizeSeqGen = Generator[
  Tuple[Optional[StorageSettings],DRef,Context,Derivation,RealizeArg],
  Tuple[Optional[L ...

realize1()

def realize1(closure: Union[Closure,Tuple[StageResult,Closure]], force_rebuild: Union[List[DRef],bool] = [], assert_realized: List[DRef] = [], realize_args: Dict[DRef,RealizeArg] = {}) -> RRef

Realize a closure, assuming that it returns a single realization.

realizeMany()

def realizeMany(closure: Union[Closure,Tuple[StageResult,Closure]], force_rebuild: Union[List[DRef],bool] = [], assert_realized: List[DRef] = [], realize_args: Dict[DRef,RealizeArg] = {}) -> List[RRef]

Realize a closure, assuming that it returns a list of realizations.

realize()

def realize(closure: Union[Closure,Tuple[StageResult,Closure]], force_rebuild: Union[List[DRef],bool] = [], assert_realized: List[DRef] = [], realize_args: Dict[DRef,RealizeArg] = {}) -> Tuple[StageResult,Closure,Context]

Takes the instantiated Closure and evaluates its targets. Calls the realizers if derivation matchers require so.

Returns the target DRefs, their closure, and the resulting Context.

See also realize1, realizeMany, repl_realize

Example:

def mystage(r:Registry)->DRef:
  ...
  return mkdrv(r, ...)

rrefs=realize1(instantiate(mystage))
print(mklen(rref).syspath)

realizeSeq()

def realizeSeq(closure: Closure, force_interrupt: List[DRef] = [], assert_realized: List[DRef] = [], realize_args: Dict[DRef,RealizeArg] = {}) -> RealizeSeqGen

realizeSeq encodes low-level details of the realization algorithm. Sequentially realize the closure by issuing steps via Python's generator interface. Consider calling realize or it's analogs instead.

FIXME: try to implement assert_realized by calling redefine with appropriate failing realizer on every Derivation.

evaluate()

def evaluate(stage, args, *,, ,, kwargs) -> RRef

Key

Key = Callable[[Optional[StorageSettings], RRef],Optional[Union[int,float,str]]]

texthash()

def texthash() -> Key

latest()

def latest() -> Key

exact()

def exact(expected: List[RRef]) -> Key

match()

def match(key: Key, trim: Callable[[List[RRef]],Optional[List[RRef]]], mnext: Optional[Matcher] = None) -> Matcher

Create a Matcher by combining different sorting keys and selecting a top-n threshold.

Only realizations which have tag 'out' (which is a default tag name) participate in matching. After the matching, Pylightnix adds all non-'out' realizations which share group with at least one matched realization.

Arguments:

  • keys: List of Key functions. Defaults ot

match_all()

def match_all(S, rrefs)

match_some()

def match_some(n: int = 1, key=None)

match_only()

def match_only()

match_latest()

def match_latest(n: int = 1) -> Matcher

match_exact()

def match_exact(rrefs: List[RRef])

cfgsp()

def cfgsp(c: Config) -> List[Tuple[str,RefPath]]

Returns the list of self-references (aka self-paths) in the config.

assert_valid_refpath()

def assert_valid_refpath(refpath: RefPath) -> None

assert_valid_config()

def assert_valid_config(c: Config) -> Config

assert_valid_name()

def assert_valid_name(s: Name) -> None

assert_valid_rref()

def assert_valid_rref(ref: str) -> None

assert_valid_hashpart()

def assert_valid_hashpart(hp: HashPart) -> None

assert_valid_dref()

def assert_valid_dref(ref: str) -> None

assert_valid_hash()

def assert_valid_hash(h: Hash) -> None

Asserts if it's Hash argument is ill-formed.

assert_valid_context()

def assert_valid_context(c: Context) -> None

assert_valid_closure()

def assert_valid_closure(closure: Closure) -> None

assert_rref_deps()

def assert_rref_deps(c: Config) -> None

assert_have_realizers()

def assert_have_realizers(r: Registry, drefs: List[DRef]) -> None

pylightnix.build

Built-in realization wrapper named Build provides helpful functions like temporary build directory management, time counting, etc.

logger

logger = getLogger(__name__)

info

info = logger.info

warning

warning = logger.warning

error

error = logger.error

BuildError Objects

def __init__(self, S: Optional[StorageSettings], dref: DRef, outpaths: Optional[Output[Path]], exception: Exception, msg: str = '')

Exception class for build errors

BuildError.__init__()

def __init__(self, S: Optional[StorageSettings], dref: DRef, outpaths: Optional[Output[Path]], exception: Exception, msg: str = '')

Initialize BuildError instance.

BuildError.__str__()

def __str__(self)

mkbuildargs()

def mkbuildargs(S: Optional[StorageSettings], dref: DRef, context: Context, starttime: Optional[str], stoptime: Optional[str], iarg: InstantiateArg, rarg: RealizeArg) -> BuildArgs

_B

_B = TypeVar('_B', bound=Build)

build_wrapper_()

def build_wrapper_(f: Callable[[_B],None], ctr: Callable[[BuildArgs],_B], nouts: Optional[int] = 1, starttime: Optional[str] = 'AUTO', stoptime: Optional[str] = 'AUTO') -> Realizer

Build Adapter which convers user-defined realizers which use Build API into a low-level Realizer

build_wrapper()

def build_wrapper(f: Callable[[Build],None], nouts: Optional[int] = 1, starttime: Optional[str] = 'AUTO', stoptime: Optional[str] = 'AUTO') -> Realizer

Build Adapter which convers user-defined realizers which use Build API into a low-level Realizer

build_config()

def build_config(b: Build) -> RConfig

Return the Config object of the realization being built.

build_context()

def build_context(b: Build) -> Context

Return the Context object of the realization being built.

build_cattrs()

def build_cattrs(b: Build) -> Any

Cache and return ConfigAttrs. Cache allows realizers to update it's value during the build process, e.g. to use it as a storage.

build_markstart()

def build_markstart(b: Build, nouts: int) -> List[Path]

build_markstop()

def build_markstop(b: Build) -> None

build_markstop_noexcept()

def build_markstop_noexcept(b: Build) -> None

rrefbstart()

def rrefbstart(rref: RRef, S=None) -> Optional[str]

Return the buildtime of the current RRef in a format specified by the PYLIGHTNIX_TIME constant.

parsetime may be used to parse stings into UNIX-Epoch seconds.

Buildtime is the time when the realization process was started. Some realizations may not provide this information.

rrefbstop()

def rrefbstop(rref: RRef, S=None) -> Optional[str]

Return the buildtime of the current RRef in a format specified by the PYLIGHTNIX_TIME constant.

parsetime may be used to parse stings into UNIX-Epoch seconds.

Buildtime is the time when the realization process was started. Some realizations may not provide this information.

rrefbdelta()

def rrefbdelta(rref: RRef, S=None) -> Optional[float]

build_outpaths()

def build_outpaths(b: Build) -> List[Path]

build_outpath()

def build_outpath(b: Build) -> Path

Return the output path of the realization being built. Output path is a path to valid temporary folder where user may put various build artifacts. Later this folder becomes a realization.

build_name()

def build_name(b: Build) -> Name

Return the name of a derivation being built.

build_deref_()

def build_deref_(b: Build, dref: DRef) -> List[RRef]

For any realization process described with it's Build handler, build_deref queries a realization of dependency dref.

build_deref is designed to be called from Realizer functions. In other cases, store_deref should be used.

build_deref()

def build_deref(b: Build, dref: DRef) -> RRef

build_paths()

def build_paths(b: Build, refpath: RefPath) -> List[Path]

build_path()

def build_path(b: Build, refpath: RefPath) -> Path

A single-realization version of the build_paths.

build_environ()

def build_environ(b: Build, env: Optional[Any] = None) -> dict

Prepare environment by adding Build's config to the environment as variables. The function resolves all singular RefPaths into system paths using current Build's context.

FIXME: Use bash-array syntax for multi-ouput paths

repl_continueBuild()

def repl_continueBuild(b: Build, rh: Optional[ReplHelper] = None) -> Optional[RRef]

repl_buildargs()

def repl_buildargs(rh: Optional[ReplHelper] = None) -> BuildArgs

repl_build()

def repl_build(rh: Optional[ReplHelper] = None, nouts: Optional[int] = 1) -> Build

Return Build object for using in repl-based debugging

Example:

from stages import some_stage, some_stage_build, some_stage_train

rh=repl_realize(instantiate(some_stage))
b=repl_build(rh)
some_stage_build(b) # Debug as needed
some_stage_train(b) # Debug as needed

repl_cancelBuild()

def repl_cancelBuild(b: Build, rh: Optional[ReplHelper] = None) -> None

pylightnix.repl

Repl module defines variants of instantiate and realize1 functions, which are suitable for REPL shells. Repl-friendly wrappers (see repl_realize) could pause the computation, save the Pylightnix state into a variable and return to the REPL's main loop. At this point user could alter the state of the whole system. Finally, repl_continue or repl_cancel could be called to either continue or cancel the realization.

ReplHelper Objects

def __init__(self, gen: RealizeSeqGen) -> None

ReplHelper.__init__()

def __init__(self, gen: RealizeSeqGen) -> None

ERR_INVALID_RH

ERR_INVALID_RH = "Neither global, nor user-defined ReplHelper is valid"

ERR_INACTIVE_RH

ERR_INACTIVE_RH = "REPL session is not paused or was already unpaused"

repl_continueAll()

def repl_continueAll(out_paths: Optional[List[Path]] = None, out_rrefs: Optional[List[RRef]] = None, rh: Optional[ReplHelper] = None) -> Optional[Context]

repl_continueMany()

def repl_continueMany(out_paths: Optional[List[Path]] = None, out_rrefs: Optional[List[RRef]] = None, rh: Optional[ReplHelper] = None) -> Optional[List[RRef]]

repl_continue()

def repl_continue(out_paths: Optional[List[Path]] = None, out_rrefs: Optional[List[RRef]] = None, rh: Optional[ReplHelper] = None) -> Optional[RRef]

repl_realize()

def repl_realize(closure: Union[Closure,Tuple[Any,Closure]], force_interrupt: Union[List[DRef],bool] = True, realize_args: Dict[DRef,RealizeArg] = {}) -> ReplHelper

TODO

Example:

rh=repl_realize(instantiate(mystage), force_interrupt=True)
# ^^^ `repl_realize` returnes the `ReplHelper` object which holds the state of
# incomplete realization
b:Build=repl_build()
# ^^^ Access it's build object. Now we may think that we are inside the
# realization function. Lets do some hacks.
with open(join(build_outpath(b),'artifact.txt'), 'w') as f:
  f.write("Fooo")
repl_continueBuild(b)
rref=repl_rref(rh)
# ^^^ Since we didn't program any other pasues, we should get the usual RRef
# holding the result of our hacks.

repl_result()

def repl_result(rh: ReplHelper) -> Optional[Context]

repl_rrefs()

def repl_rrefs(rh: ReplHelper) -> Optional[List[RRef]]

repl_rref()

def repl_rref(rh: ReplHelper) -> Optional[RRef]

repl_cancel()

def repl_cancel(rh: Optional[ReplHelper] = None) -> None

pylightnix.stages

pylightnix.stages.trivial

Trivial builtin stages

mknode()

def mknode(r: Registry, cfgdict: dict, artifacts: Dict[Name,bytes] = {}, name: str = 'mknode') -> DRef

redefine()

def redefine(stage: Any, new_config: Callable[[dict],None] = lambda x:None, new_matcher: Optional[Matcher] = None, new_realizer: Optional[Realizer] = None) -> Any

Define a new Derivation based on the existing one, by updating it's config, optionally re-writing it's matcher, or it's realizer.

Arguments:

  • stage:Any a Stage function, accepting arbitrary keyword arguments
  • new_config:Callable[[dict],None] A function to update the dref's config. Default varsion makes no changes.
  • new_matcher:Optional[Matcher]=None Optional new matcher (defaults to the existing matcher)
  • new_realizer:Optional[Realizer]=None Optional new realizer (defaults to the existing realizer)

Return: A callable Stage, accepting pass-through arguments

Example:

def _new_config(old_config):
  old_config['learning_rate'] = 1e-5
  return mkconfig(old_config)
realize1(instantiate(redefine(myMLmodel, _new_config)))

FIXME: Updating configs is dangerous: it changes its dref and thus breaks dependencies. Only top-level stages should use new_confid currently.

realized()

def realized(stage: Any) -> Stage

Asserts that the stage doesn't requre running its realizer. Re-defines stage realizer with a dummy realizer triggering an assertion.

Example:

rref:RRef=realize1(instantiate(realized(my_long_running_stage, arg="bla")))
# ^^^ Fail if `my_long_running_stage` is not yet realized.

pylightnix.stages.fetch2

Builtin stages for fetching things from the Internet

logger

logger = getLogger(__name__)

info

info = logger.info

error

error = logger.error

CURL

CURL = try_executable('curl',
                    'PYLIGHTNIX_CURL',
                    '`curl` executable ...

AUNPACK

AUNPACK = try_executable('aunpack',
                       'PYLIGHTNIX_AUNPACK',
                       '`aunp ...

fetchurl2()

def fetchurl2(url: str, sha256: Optional[str] = None, sha1: Optional[str] = None, name: Optional[str] = None, filename: Optional[str] = None, force_download: bool = False, r: Optional[Registry] = None, kwargs) -> DRef

Download file given it's URL addess.

Downloading is done by calling curl application. The path to the executable may be altered by setting the PYLIGHTNIX_CURL environment variable.

Agruments:

  • r:Registry the dependency resolution Registry.
  • url:str URL to download from. Should point to a single file.
  • sha256:str SHA-256 hash sum of the file.
  • name:Optional[str]: Name of the Derivation. The stage will attempt to deduce the name if not specified.
  • filename:Optional[str]=None Name of the filename on disk after downloading. Stage will attempt to deduced it if not specified.
  • force_download:bool=False If False, resume the last download if possible.
  • check_promises:bool=True Passed to mkdrv as-is.

Example:

def hello_src(r:Registry)->DRef:
  hello_version = '2.10'
  return fetchurl2(
    r,
    name='hello-src',
    url=f'http://ftp.gnu.org/gnu/hello/hello-{hello_version}.tar.gz',
    sha256='31e066137a962676e89f69d1b65382de95a7ef7d914b8cb956f41ea72e0f516b')

rref:RRef=realize1(instantiate(hello_src))
print(rref2path(rref))

unpack()

def unpack(path: Optional[str] = None, refpath: Optional[RefPath] = None, name: Optional[str] = None, sha256: Optional[str] = None, sha1: Optional[str] = None, aunpack_args: List[str] = [], r: Optional[Registry] = None, kwargs) -> DRef

pylightnix.stages.fetch

Builtin stages for fetching things from the Internet

logger

logger = getLogger(__name__)

info

info = logger.info

error

error = logger.error

WGET

WGET = try_executable('wget',
                    'PYLIGHTNIX_WGET',
                    'Executable `wget` ...

AUNPACK

AUNPACK = try_executable('aunpack',
                       'PYLIGHTNIX_AUNPACK',
                       '`aunp ...

_unpack_inplace()

def _unpack_inplace(o: str, fullpath: str, remove_file: bool)

fetchurl()

def fetchurl(url: str, sha256: Optional[str] = None, sha1: Optional[str] = None, mode: str = 'unpack,remove', name: Optional[str] = None, filename: Optional[str] = None, force_download: bool = False, check_promises: bool = True, r: Optional[Registry] = None, kwargs) -> DRef

Download and unpack an URL addess.

Downloading is done by calling wget application. Optional unpacking is performed with the aunpack script from atool package. sha256 defines the expected SHA-256 hashsum of the stored data. mode allows to tweak the stage's behavior: adding word 'unpack' instructs fetchurl to unpack the package, adding 'remove' instructs it to remove the archive after unpacking.

If 'unpack' is not expected, then the promise named 'out_path' is created.

Agruments:

  • r:Registry the dependency resolution Registry.
  • url:str URL to download from. Should point to a single file.
  • sha256:str SHA-256 hash sum of the file.
  • model:str='unpack,remove' Additional options. Format: [unpack[,remove]].
  • name:Optional[str]: Name of the Derivation. The stage will attempt to deduce the name if not specified.
  • filename:Optional[str]=None Name of the filename on disk after downloading. Stage will attempt to deduced it if not specified.
  • force_download:bool=False If False, resume the last download if possible.
  • check_promises:bool=True Passed to mkdrv as-is.

Example:

def hello_src(r:Registry)->DRef:
  hello_version = '2.10'
  return fetchurl(
    r,
    name='hello-src',
    url=f'http://ftp.gnu.org/gnu/hello/hello-{hello_version}.tar.gz',
    sha256='31e066137a962676e89f69d1b65382de95a7ef7d914b8cb956f41ea72e0f516b')

rref:RRef=realize1(instantiate(hello_src))
print(rref2path(rref))

fetchlocal()

def fetchlocal(sha256: str, path: Optional[str] = None, envname: Optional[str] = None, mode: str = 'unpack,remove', name: Optional[str] = None, filename: Optional[str] = None, check_promises: bool = True, r: Optional[Registry] = None, kwargs) -> DRef

Copy local file into Pylightnix storage. This function is typically intended to register application-specific files which are distributed with a source repository.

See fetchurl for arguments description.

If 'unpack' is not expected, then the promise named 'out_path' is created.

FIXME: Switch regular fetchurl to curl and call it with file:// URLs.

pylightnix.bashlike

Simple functions imitating unix shell tools.

lsdref_()

def lsdref_(r: DRef, S=None) -> Iterable[str]

lsrref_()

def lsrref_(r: RRef, fn: List[str] = [], S=None) -> Iterable[str]

lsrref()

def lsrref(r: RRef, fn: List[str] = [], S=None) -> List[str]

lsref()

def lsref(r: Union[RRef,DRef], S=None) -> List[str]

List the contents of r. For DRefs, return realization hashes. For RRefs, list artifact files.

catrref_()

def catrref_(r: RRef, fn: List[str], S=None) -> Iterable[str]

catref()

def catref(r: Union[RRef,RefPath,Path], fn: List[str] = [], S=None) -> List[str]

Return the contents of r's artifact line by line. fn is a list of folders, relative to rref's root.

rmref()

def rmref(r: Union[RRef,DRef], S=None) -> None

Forcebly remove a reference from the storage. Removing DRefs also removes all their realizations.

Currently Pylightnix makes no attempts to synchronize an access to the storage. In scenarious involving parallelization, users are expected to take care of possible race conditions.

shell()

def shell(r: Union[RRef,DRef,Build,Path,str,None] = None, S=None) -> None

Open the Unix Shell in the directory associated with the argument passed. Path to the shell executable is read from the SHELL environment variable, defaulting to /bin/sh. If r is None, open the shell in the root of the Pylightnix storage.

The function is expected to be run in REPL Python shells like IPython.

shellref()

def shellref(r: Union[RRef,DRef,None] = None, S=None) -> None

Alias for shell. Deprecated.

du()

def du(S=None) -> Dict[DRef,Tuple[int,Dict[RRef,int]]]

Calculates the disk usage, in bytes. For every derivation, return it's total disk usage and disk usages per realizations. Note, that total disk usage of a derivation is slightly bigger than sum of it's realization's usages.

find()

def find(name: Optional[Union[Stage,str]] = None, newer: Optional[float] = None, S: Optional[StorageSettings] = None) -> List[RRef]

Find RRefs in Pylightnix sotrage which match all of the criteria provided. Without arguments return all RRefs.

Arguments:

  • name:Optional[Union[Stage,str]]=None match RRefs which have name in their name. Matching is done by fnmatch Python function which supports shell-like glob expressions with '*' and '?' symbols. If name is a Stage then it is instantiated and it's name is taken.
  • newer:Optional[float]=None match RRefs which are newer than this number of seconds starting from the UNIX Epoch. Zero and negative numbers count backward from the current time.

FIXME: If name is a stage, then this function instantiates this stage before searching. Thus, the storage is moified, which may be a undesired behaviour

diff()

def diff(stageA: Union[RRef,DRef,Stage], stageB: Union[RRef,DRef,Stage], S=None) -> None

Run system's diff utility to print the difference between configs of 2 stages passed.

Note: if argument is a Stage, it is instantiated first

linkrref()

def linkrref(rref: RRef, destdir: Optional[Path] = None, format: str = '_rref_%(T)s_%(N)s', S=None) -> Path

linkkrref creates a symbolic link to a particular realization reference. The new link appears in the destdir directory if this argument is not None, otherwise the current directory is used.

Format accepts the following Python pattern tags:

  • %(T)s replaced with the build time
  • %(N)s replaced with the config name

Informally, linkrref creates the link: {tgtpath}/{format} --> $PYLIGHTNIX_STORE/{dref}/{rref}.

The function overwrites existing symlinks.

linkdref()

def linkdref(dref: DRef, destdir: Optional[Path] = None, format: str = '_rref_%(N)s', S=None) -> Path

linkrrefs()

def linkrrefs(rrefs: Iterable[RRef], destdir: Optional[Path] = None, format: str = '_rref_%(T)s_%(N)s', S=None) -> List[Path]

A Wrapper around linkrref for linking a set of RRefs.

pylightnix.lens

Lens module defines the Lens helper class, which offers quick navigation through the dependent configurations

LensContext

LensContext = NamedTuple('LensContext',[('S',Optional[StorageSettings]),
                                        ( ...

val2dict()

def val2dict(v: Any, ctx: LensContext) -> Optional[dict]

Return the dict representation of the Lens value, if possible. Getting the dictionary allows for creating new lenses

val2rref()

def val2rref(v: Any, ctx: LensContext) -> RRef

val2path()

def val2path(v: Any, ctx: LensContext) -> Path

Resolve the current value of Lens into system path. Assert if it is not possible or if the result is associated with multiple paths.

FIXME: re-use val2rref here

traverse()

def traverse(l: "Lens", hint: str) -> Any

mutate()

def mutate(l: "Lens", v: Any, hint: str) -> None

lens_repr()

def lens_repr(l, accessor: str) -> str

Lens Objects

def __init__(self, ctx: LensContext, start: Any, steps: List[str]) -> None

A Lens is a "syntactic sugar" helper object which could traverse through various Python and Pylightnix tree-like structures in a uniform way.

The list of supported structures include:

Lens lifecycle typically consists of three stages:

  1. Lens creation with mklens helper function.
  2. Navigation through the nested fileds using regular Python dot-notation. Accessing Lens's attributes results in the creation of new Lens.
  3. Access to the raw value which could no longer be converted into a Lens. In this case the raw value is returned. See val, optval, rref, dref, etc.

Lenses are not inteded to be created directly, consider using mklens constructor.

Lens.__init__()

def __init__(self, ctx: LensContext, start: Any, steps: List[str]) -> None

Arguments:

  • ctx - is the context in which this Lens is created. The more information it contains the more functions are available
  • start - Source object which we explore
  • steps - List of attributes to query

Lens.__getattr__()

def __getattr__(self, key) -> "Lens"

Sugar for Lens.get

Lens.get()

def get(self, key) -> "Lens"

Return a new Lens out of the key attribute of the current Lens

Lens.optval()

@property
def optval(self) -> Optional[Any]

Return the value of Lens as-is

Lens.val()

@val.setter
def val(self, v)

Lens.refpath()

@property
def refpath(self) -> RefPath

Check that the current value of Lens is a RefPath and return it

Lens.dref()

@property
def dref(self) -> DRef

Check that the current value of Lens is a DRef and return it

Lens.syspath()

@property
def syspath(self) -> Path

Check that the current value of Lens is a Path and return it

Lens.contents()

@property
def contents(self) -> str

Check that the current value of Lens is a Path and return it

Lens.rref()

@property
def rref(self) -> RRef

Check that the current value of Lens is an RRef and return it

Lens.closure()

@property
def closure(self) -> Closure

Constructs a closure of the DRef which this lens points to. FIXME: Filter the closure derivations from unrelated entries.

mklens()

def mklens(x: Any, o: Optional[Path] = None, b: Optional[Build] = None, rref: Optional[RRef] = None, ctx: Optional[Context] = None, closure: Optional[Closure] = None, build_output_idx: int = 0, S: Optional[StorageSettings] = None, r: Optional[Registry] = None) -> Lens

mklens creates Lens objects from various Pylightnix objects.

Arguments:

  • x:Any The object to create the Lens from. Supported source object types are:
    • RRefs
    • DRefs
    • Build
    • RefPath
    • dict
  • b:Optional[Build]=None Optional Build context of the Lens. Passing this object would allow Lens to resolve RRefs using the Context of the current realization. Also it would allow the Lens to use build_path function to resolve Build paths.
  • rref:Optional[RRef]=None Optional RRef link. Passing this object will allow Lens to resolve other RRefs using the Context of the given RRef.
  • ctx:Optional[Context]=None Passing optional Context would allow Lens to resolve RRefs.
  • build_output_idx:int=0 For Builds, specify the index of output path, defaulted to zero

Example:

stage=partial(fetchurl, url='http://example.com',
                        sha256='...',
                        output=[promise,'file.txt'],
                        foo={'bar':42}, # Additional configuration item
             )

dref:DRef=instantiate(stage).dref

mklens(dref).url.val  # Access raw value of 'url'
mklens(dref).foo             # Return another lens pointing at 'foo'
mklens(dref).foo.val         # Return raw value of 'foo' (a dict)
mklens(dref).foo.bar.val     # Return raw value of 'bar'
mklens(dref).foo.refpath     # Error! dict is not a path

mklens(dref).output.val      # Return raw output value
mklens(dref).output.refpath  # Return output as a RefPath (a list)
mklens(dref).output.syspath  # Error! not a realization

rref:RRef=realize1(instantiate(stage))

mklens(rref).output.syspath  # Return output as a system path

pylightnix.either

_REF

_REF = TypeVar('_REF')

_REF is either Path or RRef

ExceptionText

ExceptionText = str

Alias for exception text

Either Objects

def __init__(self, right: Optional[Output[_REF]] = None, left: Optional[Tuple[List[_REF],ExceptionText]] = None)

Either is a poor-man's (EitherT Exception Ouput) monad. It contains either (RIGHT) result of a realization or (LEFT) error report together with half-done results to show to the user.

Either should be considered as an "upgrade" for regular Output, which allows user to record the fact of realization failure into the special kind of realization result rather than rasing an exception.

TODO: Stress the attention on the fact that Either now encodes the list of items which may not have the same meaning. Some items may now be 'failed' and we may need a mapping function to apply matchers to them.

TODO: Think about whether we should mark the fact that a stage uses Either wrappert in the stage configuration or not. Probably we should, because either-realizers in general are not backward-compatible with regular realizers.

Either.__init__()

def __init__(self, right: Optional[Output[_REF]] = None, left: Optional[Tuple[List[_REF],ExceptionText]] = None)

mkright()

def mkright(o: Output[_REF]) -> Either[_REF]

Create a RIGHT Either

mkleft()

def mkleft(paths: List[Path], exc: ExceptionText) -> Either[Path]

Create a LEFT Either

either_paths()

def either_paths(e: Either[_REF]) -> List[_REF]

either_isRight()

def either_isRight(e: Either[_REF]) -> bool

either_isLeft()

def either_isLeft(e: Either[_REF]) -> bool

either_status()

def either_status(e: Either[_REF]) -> Optional[ExceptionText]

either_loadR()

def either_loadR(rrefs: List[RRef], S) -> Either[RRef]

either_validate()

def either_validate(dref: DRef, e: Either[Path], S=None) -> List[Path]

either_realizer()

def either_realizer(f: Callable[[Optional[StorageSettings],DRef,
                                Context,RealizeArg],Output[Path]]) -> Callable[[Optional[StorageSettings],DRef,
                                Context,RealizeArg],List[Path]]

Implements poor-man's (EitherT Exception Ouput) monad. Either, stages become either LEFT (if rasied an error) or RIGHT (after normal completion). If the stage turns LEFT, then so will be any of it's dependant stages.

Stages which use either_wrapper typically don't use claims instead of promises to allow the existance of LEFT-versions of themselves.

Either-stages should use appropriate matchers which supports LEFT-mode.

either_matcher()

def either_matcher(m: MatcherO) -> Matcher

Convert an Either-matcher into the regular Matcher

mkdrvE()

def mkdrvE(config: Config, matcher: MatcherO, realizer: RealizerO, m: Optional[Registry] = None) -> DRef

realizeE()

def realizeE(closure: Closure, force_rebuild: Union[List[DRef],bool] = [], assert_realized: List[DRef] = [], realize_args: Dict[DRef,RealizeArg] = {}) -> Either[RRef]

realizeManyE()

def realizeManyE(closure: Union[Closure,Tuple[Any,Closure]], force_rebuild: Union[List[DRef],bool] = [], assert_realized: List[DRef] = [], realize_args: Dict[DRef,RealizeArg] = {}) -> Either[RRef]

pylightnix.arch

Functions for moving parts of the storage to and from archives.

APACK

APACK = try_executable('apack',
                     'PYLIGHTNIX_APACK',
                     '`apack` execu ...

AUNPACK

AUNPACK = try_executable('aunpack',
                     'PYLIGHTNIX_AUNPACK',
                     '`aunpack` ...

spack()

def spack(roots: List[RRef], out: Path, S: Optional[StorageSettings] = None) -> None

sunpack()

def sunpack(archive: Path, S=None) -> None

deref_()

def deref_(ctxr: RRef, dref: DRef, S)

query the context for specific dref. If not present - then it must be a context holder itself.

copyclosure()

def copyclosure(rrefs_S: Iterable[RRef], S: StorageSettings, D: Optional[StorageSettings] = None) -> None

Copy the closure of rrefs from source storage S to the destination storage D. If D is None, use the default global storage as a desitnation.

TODO: Implement a non-recursive version.

pylightnix.deco

Attrs Objects

A class for proxy objects where realization config parameters are set as attributs.

unroll()

def unroll(ctx: Context, dref: DRef, b: Optional[Build], rindex: int, max_depth: Optional[int] = None, always_multiref: bool = False, S=None) -> Attrs

autodrv_()

def autodrv_(kwargs: dict, nouts: int = 1, matcher: Optional[Matcher] = None, always_multiref: bool = False, sourcedeps: List[Any] = [])

autodrv()

def autodrv(args, *,, ,, =, ,, kwargs)

autostage_()

def autostage_(nouts: int = 1, matcher: Optional[Matcher] = None, always_multiref: bool = False, sourcedeps: List[Any] = [], decokw)

autostage()

def autostage(args, *,, ,, =, ,, kwargs)

Builds a Pylightnix Stage out of a Python function. The decorator's arguments form the Configuration of a stage. After that, they go through the certain transformations and finally appear as the inner function's arguments. The transformation rules are explained in the table below.

Argument name in decorator Argument type in decorator Argument name in function Argument type in function Comment
r=None Optional[Registry] - - Registry to register this stage in
sourcedeps=[] List[Any] - - List of arbitrary Python objects to track by source in addition to the source of the wrapped function
matcher=match_latest Matcher - - Matcher to set for this stage.
always_multiref=False bool - - Set to True to represent dependencies as lists even if they include only one matched realization.
x [selfref,str,...] x str A promise to produce a file or folder
x DRef x Attrs or List[Attrs] or RRef or List[RRef] Attrs with attributs of parent realization(s) or raw Realization references
x t x t JSON-compatible arguments (bool,int,float,str,lists and dicts of thereof) are passed without changes
nouts=1 int rindex int Number of realizations to produce for this stage in one run (defaults to 1)
- - build Build Build context for the current stage

Example:

with current_registry(Registry(S)) as r:
  @autostage(a=42)
  def stage1(a,build):
    assert a==42
  @autostage(b=33,ref_stage1=stage1())
  def stage2(b,build,ref_stage1):
    assert b==33
    assert ref_stage1.a==42
  r1=realize1(instantiate(stage2))
  assert mklens(r1,S=S).b.val==33