Input Reference

The inputs Module

A collection of input stream and input rules.

class ciowarehouse2.inputs.InputStream(registry: ScriptRegistry, get_warehouse)[source]

Base class to manage input stream.

Parameters:
documents(build: Build, dbstream: DBInputStream) → list | tuple[source]

Retrieve documents.

Parameters:

  • build (cioservice.lib.build.Build) – Current build object.

  • dbstream (.models.dbinput.DBInputStream) – SqlAlchemy object to build this input stream.

Return type:

list

archive(build: Build, dbstream: DBInputStream) → tuple[Warehouse | None, CioPath | None][source]

Return the warehouse object for archive and the CioPath to the archive directory.

Parameters:

  • build (cioservice.lib.build.Build) – Current build object.

  • dbsteam (.models.dbinput.DBInputStream) – SqlAlchemy object to build this input stream.

Return type:

tuple

Returns:

A tuple such as (archive_warehouse, archive_ciopath).

archive_files(build: Build, dbstream: DBInputStream, archive_wrh: Warehouse, archive_absdir: str, input_absdir: str | None = None)[source]

Save a copy of input_absdir into archive_absdir, add archive_absdir to the repository, commit the action and refresh the archive warehouse.

Parameters:

  • build (cioservice.lib.build.Build) – Current build object.

  • dbsteam (.models.dbinput.DBInputStream) – SqlAlchemy object to build this input stream.

  • archive_wrh (.lib.warehouse.Warehouse) – Warehouse object for archive.

  • archive_absdir (str) – Absolute path to the directory for these files in the archives.

  • input_absdir (str) – (optional) Absolute path to the directory to copy.

archive_clean(build: Build, dbstream: DBInputStream, archive_wrh: Warehouse, archive_ciopath: CioPath)[source]

Loop over archives and remove old ones.

Parameters:

  • build (cioservice.lib.build.Build) – Current build object.

  • dbsteam (.models.dbinput.DBInputStream) – SqlAlchemy object to build this input stream.

  • archive_wrh (.lib.warehouse.Warehouse) – Warehouse object for archive.

  • archive_ciopath (.lib.ciopath.CioPath) – CioPath to the archive directory.

empty_input_directory(build: Build, dbstream: DBInputStream, input_wrh: Warehouse, input_absdir: str)[source]

Empty an input directory.

Parameters:

  • build (cioservice.lib.build.Build) – Current build object.

  • dbsteam (.models.dbinput.DBInputStream) – SqlAlchemy object to build this input stream.

  • input_wrh (.lib.warehouse.Warehouse) – Object for the warehouse containing the directory.

  • input_absdir (str) – Absolute path to the directory.

The inputs.stream_file Module

Class for file input stream.

class ciowarehouse2.inputs.stream_file.InputStreamFile(registry: ScriptRegistry, get_warehouse)[source]

Class to manage file input stream.

See: inputs.InputStream

documents(build: Build, dbstream: DBInputStream) → list | tuple[source]

Retrieve documents.

See: inputs.InputStream.documents()

The inputs.stream_ftp Module

Class for FTP input stream.

class ciowarehouse2.inputs.stream_ftp.InputStreamFtp(registry: ScriptRegistry, get_warehouse)[source]

Class to manage FTP input stream.

See: inputs.InputStream

documents(build: Build, dbstream: DBInputStream) → list | tuple[source]

Retrieve documents.

See: inputs.InputStream.documents()

The inputs.stream_email Module

Class for email input stream.

class ciowarehouse2.inputs.stream_email.InputStreamEmail(registry: ScriptRegistry, get_warehouse)[source]

Class to manage email input stream.

See: inputs.InputStream

documents(build: Build, dbstream: DBInputStream) → list | tuple[source]

Retrieve documents.

See: inputs.InputStream.documents()

The inputs.rule_basic Module

Class for basic input rule.

class ciowarehouse2.inputs.rule_basic.InputRuleBasic(registry: ScriptRegistry, get_warehouse, dbrule: DBInputRule)[source]

Class to manage basic input rule.

Parameters:

  • registry (chrysalio.scripts.ScriptRegistry) – Application registry.

  • get_warehouse – Function to retrieve a warehouse.

  • dbrule (.models.dbinput.DBInputRule) – SqlAlchemy object to build this rule.

self.conditions is a list of conditions. Each condition is a tuple such as (key, value_regex).

apply(build: Build, document: dict) → bool[source]

Check if the document matches with this rule and possibly send the document to its destination.

Parameters:
Return type:

bool

document2destination(build: Build, values: dict, document: dict, warehouse: Warehouse, path: str) → bool[source]

Copy the document into the destination warehouse.

Parameters:

  • build (cioservice.lib.build.Build) – Current build object.

  • values (dict) – Values of the variables.

  • document (dict) – Current document.

  • warehouse (.lib.warehouse.Warehouse) – Object for destination warehouse.

  • path (str) – Path of the document inside the warehouse.

Return type:

bool