Module Reference#

Contains references and explanations for all objects/methods in the ezycore module

ABC Classes#

BaseManager#

class ezycore.manager.BaseManager(locations: List[Union[str, BaseSegment]], models: Dict[str, Type[Model]] = {}, location_data: Dict[str, Dict[str, Any]] = {})[source]#

Standard manager class, all managers should inherit this class.

Parameters:

  • locations (List[Union[str, BaseSegment]]) – List of locations to manage

  • models (Dict[str, Model]) – Models to intialise segments with, if segment already has model, model is overwritten

  • location_data (Dict[str, Dict[str, Any]]) – Kwargs for defining segment if segment doesn’t already exist. Meaning its being passed by string

abstract add_segment(segment: Union[str, BaseSegment], **kwds) → None[source]#
Add a new segment to manager,

should raise ValueError if segment already exists

Parameters:

  • segment (Union[str, BaseSegment]) – Segment to add

  • **kwds –

    If segment being added passed as string, additional kwargs to control creation

    Note

    model kwarg MUST be provided

abstract export_segment(location: str, driver: Optional[Driver] = None, **driver_kwargs) → None[source]#

Export a segment using a driver or any class which can handle a export method

Parameters:

  • location (str) – Name of segment to export

  • driver (Driver) – Driver to use

  • **driver_kwargs – Additional kwargs for Driver.export()

get_model(location: str, *, defer: bool = False, skip_manager: bool = False) → Optional[Model][source]#
Retrieves a model being used by a segment.

If deferred, returns None if model doesn’t exist. Else raises KeyError.

Parameters:

  • location (str) – Name of segment to retrieve model from

  • defer (bool) – Whether to return None if model not found

  • skip_manager (bool) – Whether to directly search segment instead of manager

get_segment(segment: str, *, defer: bool = False) → Optional[BaseSegment][source]#

Gets a saved segment, if deferred, returns None if segment doesn’t exist. Else raises KeyError.

Parameters:

  • segment (str) – Segment to get

  • defer (bool) – Whether to return None if segment not found

models() → Iterable[Model][source]#

Returns all models assigned to manager

abstract populate(location: str, *d, data: Iterable[M] = ()) → None[source]#

Populate a segment with an array of data structs

Parameters:

  • location (str) – Name of segment to populate

  • data (Iterable[Union[dict, Segment]]) – Array of data to add

abstract populate_using_driver(location: str, driver: Driver, **driver_kwargs) → None[source]#

Populate a segment using a driver

Parameters:

  • location (str) – Name of segment to populate

  • driver (Driver) – Driver to use

  • **driver_kwargs – Additional kwargs for Driver.fetch()

abstract remove_segment(location: str, *default) → Optional[Segment][source]#
Removes an existing segment from manager,

Should raise ValueError if segment doesn’t exist, unless default is provided

Parameters:

  • location (str) – Name of segment to remove

  • *default (Any) – Value to return instead of segment

replace_model(location: str, model: Model) → None[source]#
Adds/Overwrites model to a segment.

If model is None, then segment model is removed

Parameters:

  • location (str) – Name of segment

  • model (Model) – Model to bind to segment

replace_segment(segment: Union[str, BaseSegment], new_segment: Union[str, BaseSegment]) → None[source]#

Replaces/Adds a segment regardless if segment already exists, WARNING: Inproper use may lead to unexpected errors.

  • If segment provided not found, new segment is created

  • If segment found, segment to be deleted and new segment is added

Parameters:

  • segment (Union[str, BaseSegment]) – Segment to replace

  • new_segment (Union[str, BaseSegment]) – Segment to use instead of existing segment

segments() → Iterable[BaseSegment][source]#

Returns all segments assigned to manager

abstract update_segment(location: str, **data) → BaseSegment[source]#

Updates an existing segment to handle new data

Parameters:

BaseSegment#

class ezycore.manager.BaseSegment(name: str, model: Model, *, max_size: int = 1000, make_space: bool = True)[source]#

Base class for creating segments

Parameters:

  • name (str) – Name of segment

  • max_size (int) – Maximum size of segment, if < 0 then size of segment is infinite

  • model (Model) – Model being used to store data

  • make_space (bool) – Whether to start removing content once segment is full starting from the piece of data last accessed

abstract add(obj: Union[dict, Model], *, overwrite: bool = False) → None[source]#
Adds an element within the segment,

raises ValueError if object already exists unless overwrite set to True.

Parameters:

  • obj (Union[dict, Model]) – Object to add

  • overwrite (bool) – Whether to overwrite an existing element

abstract clear() → None[source]#

Removes all elements from segment

abstract first() → Optional[Model][source]#

Retrieves the item which was most recently accessed

abstract get(obj_key: Any, *flags, default: Any = Ellipsis, **export_kwds) → Optional[Model][source]#

Retrieves an element from cache

Parameters:

  • obj_key (Any) – value to search for, set in Model._config.search_by

  • *flags – elements to include in cache, read more in the Model’s section

  • default (Any) – default value if element not found

  • **export_kwds – export kwargs, read more here

abstract invalidate_all(func: Callable[[Model], bool], *, limit: int = -1) → Iterable[Model][source]#

Invalidates all entries which match check function

Parameters:

  • func (Callable[[Model], bool]) – Function which indicates whether entry should be removed

  • limit (int) – Limit how many entries should be removed, if < 0 no limit is set

abstract keys() → Iterable[Any][source]#

Returns an iterator of all keys in segment

abstract last() → Optional[Model][source]#

Retrieves item which was last accessed

abstract pretty_print(*, limit: int = -1) → None[source]#

Prints out all data in segment, or until specified limit

Parameters:

limit (int) – How many rows to print out, starting from most request data

abstract remove(obj_key: Any, *default: Any) → Optional[Model][source]#
Removes an element from segment,

raises KeyError if key doesn’t exist unless default is provided

Parameters:

  • obj_key (Any) – Value of stored key to remove

  • *default – If value not found, returns this instead of raising an error

abstract search(func: Callable[[Model], bool], *fields, limit: int = -1, **export_kwds) → Iterable[M][source]#

Searches for elements matching query in cache

Parameters:

  • func (Callable[[Model], bool]) – Function which returns whether element is needed

  • *fields – List of fields to return from model

  • limit (int) – Number of results to restrict search to, if < 0 no limit is set.

  • **export_kwds –

    export kwargs, read more here

abstract search_using_re(expr: str, *fields, flags: int = 0, key: str = Ellipsis, limit: int = -1, **export_kwds) → Iterable[M][source]#

Searches for elements using regular expressions

Warning

Converts all values to str using the str() func

Parameters:

  • expr (str) – Regular expression to use

  • flags (int) – Flags for expression

  • *fields – List of fields to return from model

  • key (str) – Name of field to use, defaults to Config.search_by

  • limit (int) – Number of results to restrict search to, if < 0 no limit is set.

  • **export_kwds –

    export kwargs, read more here

abstract size() → int[source]#

Returns total number of elements in segment

abstract update(obj_key: Any, **kwds) → None[source]#

Updates an element in the segment

Parameters:

  • obj_key (Any) – Object key to update

  • **kwds – Fields to update

update_segment(*, name: str = Ellipsis, max_size: int = Ellipsis, model: Model = Ellipsis, make_space: bool = Ellipsis) → None[source]#

Used to update enclosed variables in a segment

Parameters:

  • name (str) – Name of segment

  • max_size (int) – Maximum size of segment, if < 0 then size of segment is infinite

  • model (Model) – Model being used to store data

  • make_space (bool) – Whether to start removing content starting from the piece of data last accessed

abstract values() → Iterable[Model][source]#

Returns an iterator of all values in segment

property make_space: bool#

Whether model should remove least accessed data

property max_size: int#

Returns maximum-size of segment

property model: Model#

Returns model of segment

property name: str#

Returns name of segment

Driver#

class ezycore.drivers.Driver[source]#

Base class for defining custom drivers

abstract export(location: str, stream: Iterator[Union[dict, Model]], include: set, exclude: set) → None[source]#

Exports data from any object supporting the __next__ method

Parameters:

  • location (str) – Location to export data to

  • stream (Iterator[Union[dict, Model]]) – An object which returns values to export

  • include (set) – set of IDs to include

  • exclude (set) – set of IDs to exclude

abstract fetch(location: str, condition: Optional[Any] = None, limit_result: int = -1, model: Optional[Model] = None, *, raw: Optional[Any] = None, no_handle: bool = False, ignore_model: bool = False, **kwds) → Iterator[Union[Dict[str, Any], Model, Tuple[Any]]][source]#

Fetches as many results as possible based of query given

Parameters:

  • location (str) – Place to fetch data from, varies between drivers.

  • condition (Any) – Filter data based on a condition, varies between drivers.

  • limit_result (int) – Limit how many results are returned, if < 0, no limit is set

  • model (Model) – Model to return data as, if no model binded to location, this model becomes the default

  • raw (Any) – Raw query data instead of using an auto generated one, allows flexibility for some drivers such SQL based drivers

  • no_handle (bool) – Whether to just return the raw fetched data

  • ignore_model (bool) – Whether to return data as dict instead of model

  • **kwds – Additional kwargs may be provided by other drivers which may require any

abstract fetch_one(location: str, condition: Optional[Any] = None, model: Optional[Model] = None, *, raw: Optional[Any] = None, no_handle: bool = False, ignore_model: bool = False, **kwds) → Optional[Union[Dict[str, Any], Model, Tuple[Any]]][source]#

Fetches only one result which matches the query

Parameters:

  • location (str) – Place to fetch data from, varies between drivers.

  • condition (Any) – Filter data based on a condition, varies between drivers.

  • limit_result (int) – Limit how many results are returned, if < 0, no limit is set

  • model (Model) – Model to return data as, if no model binded to location, this model becomes the default

  • raw (Any) – Raw query data instead of using an auto generated one, allows flexibility for some drivers such SQL based drivers

  • no_handle (bool) – Whether to just return the raw fetched data

  • ignore_model (bool) – Whether to return data as dict instead of model

  • **kwds – Additional kwargs may be provided by other drivers which may require any

abstract map_to_model(**kwds) → None[source]#

Maps locations to internal data spots.

e.g. Segment may be called users, but SQL table may be called user_table

Parameters:

**kwds –

Keyword arguments for mapping from in-program name to database name.

py
d = Driver(..., models={'users': UserModel})
d.map_to_model(users='my_table')

Default Implementations#

Manager#

class ezycore.manager.Manager(locations: List[Union[str, BaseSegment]], models: Dict[str, Type[Model]] = {}, location_data: Dict[str, Dict[str, Any]] = {})[source]#

Default manager class implementation

Parameters:

  • locations (List[Union[str, BaseSegment]]) – List of locations to manage

  • models (Dict[str, Model]) – Models to intialise segments with, if segment already has model, model is overwritten

  • location_data (Dict[str, Dict[str, Any]]) – Kwargs for defining segment if segment doesn’t already exist. Meaning its being passed by string

add_segment(segment: Union[str, Segment], **kwds) → None[source]#
Add a new segment to manager,

should raise ValueError if segment already exists

Parameters:

  • segment (Union[str, BaseSegment]) – Segment to add

  • **kwds –

    If segment being added passed as string, additional kwargs to control creation

    Note

    model kwarg MUST be provided

export_segment(location: str, driver: Optional[Driver] = None, **driver_kwargs) → None[source]#

Export a segment using a driver or any class which can handle a export method

Parameters:

  • location (str) – Name of segment to export

  • driver (Driver) – Driver to use

  • **driver_kwargs – Additional kwargs for Driver.export()

get_model(location: str, *, defer: bool = False, skip_manager: bool = False) → Optional[Model]#
Retrieves a model being used by a segment.

If deferred, returns None if model doesn’t exist. Else raises KeyError.

Parameters:

  • location (str) – Name of segment to retrieve model from

  • defer (bool) – Whether to return None if model not found

  • skip_manager (bool) – Whether to directly search segment instead of manager

get_segment(segment: str, *, defer: bool = False) → Optional[BaseSegment]#

Gets a saved segment, if deferred, returns None if segment doesn’t exist. Else raises KeyError.

Parameters:

  • segment (str) – Segment to get

  • defer (bool) – Whether to return None if segment not found

models() → Iterable[Model]#

Returns all models assigned to manager

populate(location: str, *d, data: Iterable[M] = ()) → None[source]#

Populate a segment with an array of data structs

Parameters:

  • location (str) – Name of segment to populate

  • data (Iterable[Union[dict, Segment]]) – Array of data to add

populate_using_driver(location: str, driver: Driver, **driver_kwargs) → None[source]#

Populate a segment using a driver

Parameters:

  • location (str) – Name of segment to populate

  • driver (Driver) – Driver to use

  • **driver_kwargs – Additional kwargs for Driver.fetch()

remove_segment(location: str, *default) → Optional[Segment][source]#
Removes an existing segment from manager,

Should raise ValueError if segment doesn’t exist, unless default is provided

Parameters:

  • location (str) – Name of segment to remove

  • *default (Any) – Value to return instead of segment

replace_model(location: str, model: Model) → None#
Adds/Overwrites model to a segment.

If model is None, then segment model is removed

Parameters:

  • location (str) – Name of segment

  • model (Model) – Model to bind to segment

replace_segment(segment: Union[str, BaseSegment], new_segment: Union[str, BaseSegment]) → None#

Replaces/Adds a segment regardless if segment already exists, WARNING: Inproper use may lead to unexpected errors.

  • If segment provided not found, new segment is created

  • If segment found, segment to be deleted and new segment is added

Parameters:

  • segment (Union[str, BaseSegment]) – Segment to replace

  • new_segment (Union[str, BaseSegment]) – Segment to use instead of existing segment

segments() → Iterable[BaseSegment]#

Returns all segments assigned to manager

update_segment(location: str, **data) → Segment[source]#

Updates an existing segment to handle new data

Parameters:

Segment#

class ezycore.manager.Segment(name: str, model: Model, *, max_size: int = 1000, make_space: bool = True)[source]#

Default segment class

Parameters:

  • name (str) – Name of segment

  • max_size (int) – Maximum size of segment

  • model (Model) – Model being used to store data

  • make_space (bool) – Whether to start removing content starting from the piece of data last accessed

add(obj: M, *, overwrite: bool = False) → None[source]#
Adds an element within the segment,

raises ValueError if object already exists unless overwrite set to True.

Parameters:

  • obj (Union[dict, Model]) – Object to add

  • overwrite (bool) – Whether to overwrite an existing element

clear() → None[source]#

Removes all elements from segment

first() → Optional[Model][source]#

Retrieves the item which was most recently accessed

get(obj_key: Any, *flags, default: Any = Ellipsis, **export_kwds) → Optional[Model][source]#

Retrieves an element from cache

Parameters:

  • obj_key (Any) – value to search for, set in Model._config.search_by

  • *flags – elements to include in cache, read more in the Model’s section

  • default (Any) – default value if element not found

  • **export_kwds –

    export kwargs, read more here

invalidate_all(func: Callable[[Model], bool], *, limit: int = -1) → Iterable[Model][source]#

Invalidates all entries which match check function

Parameters:

  • func (Callable[[Model], bool]) – Function which indicates whether entry should be removed

  • limit (int) – Limit how many entries should be removed, if < 0 no limit is set

keys() → Iterable[Any][source]#

Returns an iterator of all keys in segment

last() → Optional[Model][source]#

Retrieves item which was last accessed

newest(limit: int = -1) → Iterable[Model][source]#

Retrieves elements starting from the most recently accessed values

Parameters:

limit (int) – How many elements to retreive, if < 0 then all elements are retrieved

oldest(limit: int = -1) → Iterable[Model][source]#

Retrieves elements starting from the least accessed values

Parameters:

limit (int) – How many elements to retrieve, if < 0 then all elements are retrieved

pretty_print(*, limit: int = -1) → None[source]#

Prints out all data in segment, or until specified limit

Parameters:

limit (int) – How many rows to print out, starting from most request data

remove(obj_key: Any, *default: Any) → Optional[Model][source]#
Removes an element from segment,

raises KeyError if key doesn’t exist unless default is provided

Parameters:

  • obj_key (Any) – Value of stored key to remove

  • *default – If value not found, returns this instead of raising an error

search(func: Callable[[Model], bool], *fields, limit: int = -1, **export_kwds) → Iterable[M][source]#

Searches for elements matching query in cache

Parameters:

  • func (Callable[[Model], bool]) – Function which returns whether element is needed

  • *fields – List of fields to return from model

  • limit (int) – Number of results to restrict search to, if < 0 no limit is set.

  • **export_kwds –

    export kwargs, read more here

search_using_re(expr: str, *fields, flags: int = 0, key: Optional[str] = None, limit: int = -1, **export_kwds) → Iterable[M][source]#

Searches for elements using regular expressions

Warning

Converts all values to str using the str() func

Parameters:

  • expr (str) – Regular expression to use

  • flags (int) – Flags for expression

  • *fields – List of fields to return from model

  • key (str) – Name of field to use, defaults to Config.search_by

  • limit (int) – Number of results to restrict search to, if < 0 no limit is set.

  • **export_kwds –

    export kwargs, read more here

size() → int[source]#

Returns total number of elements in segment

update(obj_key: Any, **kwds) → None[source]#

Updates an element in the segment

Parameters:

  • obj_key (Any) – Object key to update

  • **kwds – Fields to update

update_segment(*, name: str = Ellipsis, max_size: int = Ellipsis, model: Model = Ellipsis, make_space: bool = Ellipsis) → None#

Used to update enclosed variables in a segment

Parameters:

  • name (str) – Name of segment

  • max_size (int) – Maximum size of segment, if < 0 then size of segment is infinite

  • model (Model) – Model being used to store data

  • make_space (bool) – Whether to start removing content starting from the piece of data last accessed

values() → Iterable[Model][source]#

Returns an iterator of all values in segment

property make_space: bool#

Whether model should remove least accessed data

property max_size: int#

Returns maximum-size of segment

property model: Model#

Returns model of segment

property name: str#

Returns name of segment

SQLiteDriver#

class ezycore.drivers.SQLiteDriver(database: ~typing.Union[str, bytes, ~os.PathLike[str], ~os.PathLike[bytes]], timeout: float = 5.0, detect_types: int = 0, isolation_level: ~typing.Optional[str] = 'DEFERRED', check_same_thread: bool = True, factory: ~typing.Optional[~typing.Type[~sqlite3.Connection]] = <class 'sqlite3.Connection'>, cached_statements: int = 128, uri: bool = False, cursorClass: ~typing.Optional[~typing.Any] = None, models: ~typing.Dict[str, ~ezycore.models.core.Model] = {}, model_maps: ~typing.Dict[str, str] = {})[source]#

Default implementation for the SQLite driver

Note

All paramaters not listed in the Parameters section may be found here, https://docs.python.org/3/library/sqlite3.html#sqlite3.connect

Parameters:

  • models (Dict[str, Model]) – Models to convert fetched results to, mapping must be table name to model

  • model_maps (Dict[str, str]) – Mapping from model key to database table name

export(location: str, stream: Iterator[Union[dict, Model]]) → None[source]#

Exports data from any object supporting the __next__ method

Parameters:

  • location (str) – Location to export data to

  • stream (Iterator[Union[dict, Model]]) – An object which returns values to export

  • include (set) – set of IDs to include

  • exclude (set) – set of IDs to exclude

fetch(location: str, condition: str = '', limit_result: int = -1, model: Optional[Model] = None, *, raw: Optional[str] = None, no_handle: bool = False, ignore_model: bool = False, parameters: Tuple[Any] = ()) → Optional[Iterator[Union[Dict[str, Any], Model, Tuple[Any]]]][source]#

Fetches data from a table

Parameters:

  • location (str) – Table to fetch data from

  • condition (Any) – Condition to use in WHERE statement, provide arg without the WHERE clause.

  • limit_result (int) – Limit how many results are returned, if < 0, no limit is set

  • model (Model) – Model to return data as, if no model binded to location, this model becomes the default

  • raw (Any) – Raw query data instead of using an auto generated one, allows flexibility for some drivers such SQL based drivers

  • no_handle (bool) – Whether to just return the raw fetched data

  • ignore_model (bool) – Whether to return data as dict instead of model

  • parameters (Tuple[Any]) – Parameters to be supplied with statement, flexibility if using the raw arg

fetch_one(location: str, condition: Optional[Any] = None, model: Optional[Model] = None, *, raw: Optional[Any] = None, no_handle: bool = False, ignore_model: bool = False, parameters: Tuple[Any] = ()) → Optional[Union[Dict[str, Any], Model, Tuple[Any]]][source]#

Fetches only 1 item

map_to_model(**kwds) → None[source]#

Maps locations to internal data spots.

e.g. Segment may be called users, but SQL table may be called user_table

Parameters:

**kwds –

Keyword arguments for mapping from in-program name to database name.

py
d = Driver(..., models={'users': UserModel})
d.map_to_model(users='my_table')