atomdb.base module¶

Distributed under the terms of the MIT License.

The full license is in the file LICENSE.text, distributed with this software.

Created on Jun 12, 2018

class atomdb.base.JSONModel[source]¶

Bases: atomdb.base.Model

A simple model that can be serialized to json. Useful for embedding within other models.

objects: ClassVar[atomdb.base.ModelManager]¶: Handles database access. Subclasses should redefine this.

serializer: ClassVar[atomdb.base.ModelSerializer] = <atomdb.base.JSONSerializer object>¶: Handles encoding and decoding. Subclasses should redefine this to a subclass of ModelSerializer

class atomdb.base.JSONSerializer[source]¶

Bases: atomdb.base.ModelSerializer

flatten(v: Any, scope: Optional[Dict[int, Any]] = None)[source]¶: Flatten date, datetime, time, decimal, and bytes as a dict with a __py__ field and arguments to reconstruct it. Also see the coercers

flatten_object(obj: atomdb.base.Model, scope: Dict[int, Any]) → Dict[str, Any][source]¶: Flatten to just json but add in keys to know how to restore it.

async get_object_state(obj: Any, state: Dict[str, Any], scope: Dict[int, Any])[source]¶: State should be contained in the dict

registry¶: Store all registered models

class atomdb.base.Model[source]¶

Bases: atom.atom.Atom

An atom model that can be serialized and deserialized to and from a database.

async delete()[source]¶: Alias to delete this object in the database

async load()[source]¶: Alias to load this object from the database

objects: ClassVar[atomdb.base.ModelManager]¶: Handles database access. Subclasses should redefine this.

async classmethod restore(state: Dict[str, Any], **kwargs: Any) → atomdb.base.M[source]¶: Restore an object from the database state

async save()[source]¶: Alias to delete this object to the database

serializer: ClassVar[atomdb.base.ModelSerializer] = <atomdb.base.ModelSerializer object>¶: Handles encoding and decoding. Subclasses should redefine this to a subclass of ModelSerializer

class atomdb.base.ModelManager[source]¶

Bases: atom.atom.Atom

A descriptor so you can use this somewhat like Django’s models. Assuming your using motor.

MyModel.objects.find_one({‘_id’:’someid})

database¶: Used to access the database

classmethod instance() → atomdb.base.ModelManager[source]¶

class atomdb.base.ModelMeta(name, bases, dct)[source]¶: Bases: atom.atom.AtomMeta

class atomdb.base.ModelSerializer[source]¶

Bases: atom.atom.Atom

Handles serializing and deserializing of Model subclasses. It will automatically save and restore references where present.

coercers¶: Mapping of type name to coercer function

flatten(v: Any, scope: Optional[Dict[int, Any]] = None) → Any[source]¶

Convert Model objects to a dict

v: Object: The object to flatten
scope: Dict: The scope of references available for circular lookups

result: Object: The flattened object

flatten_object(obj: atomdb.base.Model, scope: Dict[int, Any]) → Any[source]¶

Serialize a model for entering into the database

obj: Model: The object to unflatten
scope: Dict: The scope of references available for circular lookups

result: Object: The flattened object

async get_object_state(obj: atomdb.base.Model, state: Any, scope: Dict[int, Any]) → Any[source]¶

Lookup the state needed to restore the given object id and class.

obj: Model: The object created by get_or_create
state: Dict: Unflattened state of object to restore
scope: Dict: Scope of objects available when flattened

result: Any: The model state needed to restore this object

async get_or_create(cls: Type[atomdb.base.Model], state: Any, scope: Dict[int, Any]) → Tuple[atomdb.base.Model, bool][source]¶

Get a cached object for this _id or create a new one. Subclasses should override this as needed to provide object caching if desired.

cls: Class: The type of object expected
state: Dict: Unflattened state of object to restore
scope: Dict: Scope of objects available when flattened

result: Tuple[object, bool]: A tuple of the object and a flag stating if it was created or not.

classmethod instance() → atomdb.base.ModelSerializer[source]¶

registry¶: Store all registered models

async unflatten(v: Any, scope: Optional[Dict[int, Any]] = None) → Any[source]¶

Convert dict or list to Models

v: Dict or List: The object(s) to unflatten
scope: Dict: The scope of references available for circular lookups

result: Object: The unflattened object

async unflatten_object(cls: Type[atomdb.base.Model], state: Dict[str, Any], scope: Dict[int, Any]) → Optional[atomdb.base.Model][source]¶

Restore the object for the given class, state, and scope. If a reference is given the scope should be updated with the newly created object using the given ref.

cls: Class: The type of object expected
state: Dict: The state of the object to restore

result: object or None: A the newly created object (or an existing object if using a cache) or None if this object does not exist in the database.

exception atomdb.base.UnresolvableError(member)[source]¶

Bases: Exception

Error raised when a Forwarded Member cannot be resolved at the time when the resolve_member_types is called.

atomdb.base.find_subclasses(cls: Type[atomdb.base.T]) → List[Type[atomdb.base.T]][source]¶: Finds subclasses of the given class

atomdb.base.generate_function(source: str, namespace: Dict[str, Any], fn_name: str, optimize: bool = True) → Callable[[...], Any][source]¶

Generate an optimized function

source: str: The function source code
namespaced: dict: Namespace available to the function
fn_name: str: The name of the generated function.

fn: function: The function generated.

atomdb.base.generate_getstate(cls: Type[atomdb.base.Model]) → Callable[[atomdb.base.M, Optional[Dict[int, Any]]], Dict[str, Any]][source]¶

Generate an optimized __getstate__ function for the given model.

cls: Type[Model]: The clase to generate a getstate function for.

result: GetStateFn: A function optimized to generate the state for the given model class.

atomdb.base.generate_restorestate(cls: Type[atomdb.base.Model]) → Callable[[atomdb.base.M, Dict[str, Any], Optional[Dict[int, Any]]], None][source]¶

Generate an optimized __restorestate__ function for the given model.

cls: Type[Model]: The clase to generate a getstate function for.

result: RestoreStateFn: A function optimized to restore the state for the given model class.

atomdb.base.getrandbits(k) → x. Generates an int with k random bits.¶

atomdb.base.is_db_field(m: atom.catom.Member) → bool[source]¶

Check if the member should be saved into the database. Any member that does not start with an underscore, is not a Property, and is not tagged with store=False is considered to be field to save into the database.

m: Member: The atom member to check.

result: bool: Whether the member should be saved into the database.

atomdb.base.is_primitive_member(m: atom.catom.Member) → Optional[bool][source]¶

Check if the member can be serialized without calling flatten. If the member references a field that is not yet resolved it returns None indicating that it cannot determine whether it is primitive yet.

m: Member: The atom member to check.

result: Optional[bool]: Whether the member is a primitive type that can be intrinsicly converted.

atomdb.base.resolve_member_types(member: atom.catom.Member, resolve: bool = True) → Optional[Tuple[type, ...]][source]¶

Determine the validation types specified on a member.

member: Member: The member to retrieve the type from
resolve: bool: Whether to resolve “Forward” members.

types: Optional[Tuple[Model|Member|type, ..]]: The member types. If types is None then the member does not do any type validation.

UnresolveableError: If resolve=False and the member has a nested forwarded member this will raise an UnresolvableError with the unresolved member.