tethys.core.regobjs.regobj_zero

Module Contents

tethys.core.regobjs.regobj_zero.RegObjZT

tethys.core.regobjs.regobj_zero.log

tethys.core.regobjs.regobj_zero.CLASS_PATH_PREFIX = /zero/

class tethys.core.regobjs.regobj_zero.ZeroRegistrableObject

Bases: tethys.core.regobjs.regobj_base.RegistrableObjectBase

Base class for all entities that used like models for a database. It’s the Zero generation.

A few examples:

class SomeClass(ZeroRegistrableObject):
    FIELDS_SCHEMA = {
        "some_field": {"type": "string"}
    }

    def __init__(self, some_field, another_field, **kwargs):
        self.some_field = some_field
        self.another_field = another_field

    # ...

# -------------------------------------------------

obj1 = SomeClass("str", "no:save", _id="some_id")
obj2 = SomeClass.load("some_id")

assert id(obj1) == id(obj2)

# -------------------------------------------------

obj1 = SomeClass("str", "no:save")
obj1.save()

obj2 = ZeroRegistrableObject.load(obj1.path)

assert id(obj1) == id(obj2)

# -------------------------------------------------

ZeroRegistrableObject.REPOSITORY = JsonFileRepository(create=True)

obj1 = SomeClass(some_field="str", another_field="no:save", _id="some_id")
obj1.save()

assert obj1.some_field == "str"
assert obj1.another_field == "no:save"

# restart script

obj2 = SomeClass.load("some_id")
assert obj1.some_field == "str"
assert obj1.another_field == "no:save"  # raise AttributeError: 'SomeClass' object has no attribute 'another_field'

Create and return a new object. See help(type) for accurate signature.

CLASS_PATH = /sandbox/

Default collection path in the repository

FIELDS_SCHEMA

Default fields validators

REPOSITORY

Entities repository. You can override it in your own classes

__getnewargs_ex__(self)

__getstate__(self) → dict

__setstate__(self, state: dict)

classmethod prepare(cls, item: dict, keys: list = None) → dict

Validate and normalize data in the attrs that define in the FIELDS_SCHEMA

classmethod generate_id(cls) → str

Generate random ID. Default: uuid4

classmethod generate_path(cls, obj_id: str) → str

Generate collection path from the ID

property id(self) → str

Return self._id value

property path(self) → str

Return generate_path(ID) result

property version(self) → str

Return generate_version() result

generate_version(self) → str

Return hash of the fields data

__getattr__(self, item: str)

__setattr__(self, key: str, value: Any)

Implement setattr(self, name, value).

__str__(self)

Return str(self).

__hash__(self)

Return hash(self).

__eq__(self, other: Any)

Return self==value.

copy(self)

classmethod load(cls: Type[RegObjZT], obj_id: str, with_cache: bool = True, **kwargs) → RegObjZT

Load the entity from the repository by ID. If the entity already in the memory then method will return cached version.

Parameters

• obj_id (str) – Entity ID or Entity path

• with_cache (bool) – load data from the cache (default: True)

Returns

ZeroRegistrableObject instance

Return type

ZeroRegistrableObject

classmethod list(cls: Type[RegObjZT], list_filter: dict = None, **kwargs) → List[RegObjZT]

Load list of the entities. You can specify some filters, but the filter’s query language is experimental.

Filter example:

{“some_field>=”: 2} - list entities where some_field >= 2

“{field_name}{operator}” - key template

Basic filters:

operator	python op.
->	field in {value}
!>	field not in {value}
>=	field >= {value}
>	field > {value}
<=	field <= {value}
<	field < {value}
!=	field != {value}
==	field == {value}

Parameters

list_filter (dict) – dictionary of the filters

Returns

list of the ZeroRegistrableObject instances

Return type

List(ZeroRegistrableObject)

refresh(self: RegObjZT, ignore_errors: bool = True, **kwargs) → Optional[RegObjZT]

Reload entity from the repository (without cache)

Parameters

ignore_errors (bool) – ignore errors like TethysRONotFound (default: True)

Returns

return ZeroRegistrableObject instance or None (when error like NotFound)

Return type

ZeroRegistrableObject or None

save(self: RegObjZT, save_dependency: bool = True, save_dependency_depth: int = 6, save_dependency_search_depth: int = 0, **kwargs) → RegObjZT

Save current state to the repository.

Parameters

• save_dependency (bool) – save related entities

• save_dependency_depth (int) – depth of related entities recursion (-1 = infinity)

• save_dependency_search_depth – depth of searching in the fields values (like dict).

delete(self: RegObjZT, **kwargs) → RegObjZT

Delete the object from the repository.

Returns

return self object

Return type

ZeroRegistrableObject

lock(self, lock_ttl: float = 60, wait_timeout: float = float('inf'), blocking: bool = True, **kwargs) → bool

Lock the object in the repository.

Parameters

• lock_ttl (float) – time to live for the lock (next lock will wait for the time or wait_timeout)

• wait_timeout (float) – how much time to wait until lock is unlocked (after this time the function will have ignored the lock)

• blocking (bool) – wait for the lock (if false then the function return False)

Returns

is locked by the current process?

unlock(self, **kwargs) → bool

Unlock the object in the repository.

lock_context(self, lock_ttl: float = 60, wait_timeout: float = float('inf'), blocking: bool = True, **kwargs)

Contextmanager for the Locking/Unlocking the object in the repository. It allows nested context

Example:

start = time.time()

def lock():
    obj.lock()
    print(time.time() - start)  # ~ 2s

with obj.lock_context(lock_ttl=1):
    with obj.lock_context(lock_ttl=2):
        print(time.time() - start)  # ~ 0s

        Thread(target=lock).start()
        time.sleep(5)  # todo_something