cgnal.core.data.layer package

Subpackages

Module contents

Data layer module.

class cgnal.core.data.layer.Archiver(*args, **kwds)

Bases: Generic[cgnal.core.typing.T], abc.ABC

Object that retrieve data from source and stores it in memory.

abstract archive(obj: cgnal.core.typing.T) → Union[cgnal.core.data.layer.Archiver, None, pymongo.results.UpdateResult, List[pymongo.results.UpdateResult]]

Archive an object in the persistance layer.

Parameters: obj – object to be archived
Returns: archiver or update result

foreach(f: Callable[[cgnal.core.typing.T], None], *args, **kwargs) → None

Apply a function to each object.

Parameters

f – function
args – other arguments passed to the retrieve method
kwargs – other keyworded arguments passed to the retrieve method

map(f: Callable[[cgnal.core.typing.T], cgnal.core.data.layer.V], *args: Any, **kwargs: Any) → Iterator[cgnal.core.data.layer.V]

Apply a tranformation to each object and return an iterator.

Parameters

f – function
args – other arguments passed to the retrieve method
kwargs – other keyworded arguments passed to the retrieve method

Yield

iterator over transformed objects

abstract retrieve(condition: Any, sort_by: Any) → Iterator[cgnal.core.typing.T]

Retrieve objects under given filter conditions and sorting options.

Parameters

condition – filter to be applied in the persistance layer
sort_by – sorting options

Yield

iterator over the objects retrieved from the persistance layer

retrieveGenerator(condition: Any, sort_by: Any) → cgnal.core.data.model.core.IterGenerator[cgnal.core.typing.T]

Wrap the retrieve method to return an iterator generator.

Parameters

condition – filter condition
sort_by – sorting options

Returns

iterator generator

class cgnal.core.data.layer.DAO(*args, **kwds)

Bases: Generic[cgnal.core.typing.T, cgnal.core.data.layer.V], abc.ABC

Data Access Object.

abstract computeKey(obj: cgnal.core.typing.T) → Union[Hashable, Dict[str, bson.objectid.ObjectId]]

Compute the key of a given object.

Parameters: obj – object
Returns: hashable key

abstract get(obj: cgnal.core.typing.T) → cgnal.core.data.layer.V

Return the record from the persistance layer corresponding to a object.

Parameters: obj – object
Returns: value

abstract parse(row: cgnal.core.data.layer.V) → cgnal.core.typing.T

Parse the record of the persistance layer into a object.

Parameters: row – record of the persistance layer
Returns: object

class cgnal.core.data.layer.DatabaseABC

Bases: abc.ABC

Abstract class for databases.

abstract table(table_name: str) → Optional[cgnal.core.data.layer.TableABC]

Return the table with a given name.

Parameters: table_name – name of the table
Returns: table or None if table not found

class cgnal.core.data.layer.EmptyDatabase

Bases: cgnal.core.data.layer.DatabaseABC

Class for empty Databases.

table(table_name: str) → None

Raise an exception, since the database is empty.

Parameters: table_name – name of the table
Raises: NoTableException – no table is present in an empty database

class cgnal.core.data.layer.TableABC

Bases: abc.ABC

Abstract class for tables.

abstract to_df(query: str) → pandas.core.frame.DataFrame

Transform the underling data into a pandas dataframe.

Parameters: query – query
Returns: pandas dataframe

abstract write(df: pandas.core.frame.DataFrame) → None

Write the pandas dataframe in the persistance layer.

Parameters: df – pandas dataframe

class cgnal.core.data.layer.Writer

Bases: abc.ABC

Abstract class to write Tables.

abstract push(df: pandas.core.frame.DataFrame) → None

Push a pandas dataframe.

Parameters: df – dataframe

abstract property table: cgnal.core.data.layer.TableABC: Return table object.