ddb_single package¶

Submodules¶

ddb_single.error module¶

exception ddb_single.error.DDBSingleError(message)¶: Bases: Exception

exception ddb_single.error.InvalidParameterError(message)¶: Bases: DDBSingleError

exception ddb_single.error.NotFoundError(message)¶: Bases: DDBSingleError

exception ddb_single.error.ValidationError(message)¶: Bases: DDBSingleError

ddb_single.model module¶

class ddb_single.model.BaseModel(__skip_validation__=False, **kwargs)¶

Bases: object

The base model class. .. rubric:: Examples

>>> class User(BaseModel):
...     __table__ = table
...     __model_name__ = "user"
...     id = DBField(primary_key=True, required=True)
...     name = DBField(required=True)
...     age = DBField(type=FieldType.NUMBER)

classmethod describe()¶

Returns:: The description of the model.
Return type:: dict

get_field(key: str) → DBField¶

Parameters:: key – DBField name.
Returns:: DBField instance.
Return type:: DBField

class ddb_single.model.DBField(type: FieldType = FieldType.STRING, default=None, default_factory=None, nullable=True, required=False, primary_key=False, secondary_key=False, unique_key=False, search_key=False, reletion=None, reletion_by_unique=True, relation_raise_if_not_found=False, ignore_case=False, **kwargs)¶

Bases: object

DBField is a field of a model. It is used to define the structure of a model.

begins_with(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.BEGINS_WITH.
Return type:: dict

between(value1, value2)¶

Parameters:

value1 – The value to be compared.
value2 – The value to be compared.

Returns:

key_ex with mode QueryType.BETWEEN.

Return type:

dict

contains(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.CONTAINS.
Return type:: dict

describe()¶

eq(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.EQ.
Return type:: dict

gt(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.GT.
Return type:: dict

gte(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.GTE.
Return type:: dict

in_(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.IN.
Return type:: dict

is_list()¶

Returns:: Whether the field is a list.
Return type:: bool

key_ex(value, mode)¶

Parameters:

value – The value to be compared.
mode – The mode of the comparison.

Returns:

The key expression.

Return type:

dict

lt(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.LT.
Return type:: dict

lte(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.LTE.
Return type:: dict

ne(value)¶

Parameters:: value – The value to be compared.
Returns:: key_ex with mode QueryType.NE.
Return type:: dict

search_data_key()¶

Returns:: The search data key of the field.
Return type:: str

search_item(pk)¶

Parameters:: pk – The primary key of the item.
Returns:: The search item.
Return type:: dict

search_key_factory()¶

Returns:: The search key of the field.
Return type:: str

serch_index()¶

Returns:: The search index of the field.
Return type:: str

setup(name, model_cls)¶

validate(value=None, skip=False)¶

Parameters:: value – The value to be validated. If not provided, the value of the field will be used.
Returns:: The validated value.

ddb_single.query module¶

class ddb_single.query.Query(table: Table, model: BaseModel | None = None)¶

Bases: object

create(batch=None, raise_if_exists=False)¶: Create a new item. :param batch: BatchWriteItem :param raise_if_exists: Throw ValidationError if item already exists :type raise_if_exists: bool

delete(target: dict = None, batch=None)¶: Delete an item. :param batch: BatchWriteItem

delete_by_pk(pk, batch=None)¶: Delete an item by primary key. :param pk: Primary key :param batch: BatchWriteItem

delete_by_unique(value, batch=None)¶: Delete an item by unique key. :param batch: BatchWriteItem

get(pk: str)¶: Get single item. :param pk: Primary key

get_by_unique(value, pk_only=False, keys: List[str | DBField] = [])¶: Get single item by unique key. :param value: Unique key value :param keys: Unique key. If not specified, search by all unique keys.

get_pk()¶

get_reference(model: BaseModel = '', field: DBField = '', pk_only=False)¶

get_relation(model: BaseModel = '', field: DBField = '', pk_only=False)¶

get_unique()¶

model(model: BaseModel)¶: Set DB data model. :param model: DB data model :type model: BaseModel

search(*queries, pk_only=False, limit=None)¶: Search items. :param queries: Query objects :type queries: List[dict]

update(target: dict = None, batch=None)¶: Update an item. :param target: Target item :param batch: BatchWriteItem

validate()¶

ddb_single.table module¶

class ddb_single.table.FieldType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)¶

Bases: Enum

BINARY = 'B'¶

BINARY_SET = 'BS'¶

BOOLEAN = 'BOOL'¶

LIST = 'L'¶

MAP = 'M'¶

NULL = 'NULL'¶

NUMBER = 'N'¶

NUMBER_SET = 'NS'¶

STRING = 'S'¶

STRING_SET = 'SS'¶

class ddb_single.table.Table(table_name, primary_key='pk', primary_key_type='S', primary_key_factory=<function default_pk_factory>, primary_key2model=<function default_pk2model>, range_index_name='RangeIndex', secondary_key='sk', secondary_key_type='S', secondary_key_prefix='', secondary_key_suffix='_item', secondary_key_factory=<function default_sk_factory>, unique_key='unique', unique_key_type='S', search_prefix='search_', relation_prefix='rel_', relation_joint='', search_data_key='data', search_index='DataSearchIndex', search_data_num_key='data-n', search_num_index='NumDataSearchIndex', search_data_bin_key='data-b', search_bin_index='BinDataSearchIndex', ReadCapacityUnits=1, WriteCapacityUnits=1, **table_kwargs)¶

Bases: object

all_items(pk_only=False) → list[dict]¶: dump all items in the table

batch_create(items, batch=None)¶

batch_delete_items(items, batch=None)¶

batch_get(keys)¶

batch_get_from_pks(pks: list) → list¶

batch_update(items, batch=None)¶

batch_writer()¶

clear_item(pk, batch=None)¶

clear_reference(pk, model_name='', batch=None)¶

clear_relation(pk, model_name='', batch=None)¶

create(item, batch=None)¶

create_table()¶

delete(pk, sk, batch=None)¶

detele_item(item, batch=None)¶

filter(items, searchExs)¶

get_item(pk, sk=None)¶

init(create_if_not_exists=True)¶

list_models() → list[dict]¶: Show tables in the database :returns: table information. e.g. [{“table_name”: “table1”, “count”: 10}, …] :rtype: list[dict]

pk(model_name)¶

pk2model(pk)¶

pk2rel_key(pk)¶

pk2sk(pk)¶

reference(pk, model_name='', field_name='', pk_only=False)¶

rel_key(pk)¶

rel_key2pk(rel_key)¶

rel_prefix(model_name='')¶

relation(pk, model_name='', field_name='', pk_only=False)¶

search(model_name, *searchEx, pk_only=False, limit=None)¶

Search items :param model_name: model name :type model_name: str :param searchEx: search expression :type searchEx: list[dict] :param pk_only: return only primary key. Defaults to False. :type pk_only: bool, optional :param limit: limit. Defaults to None. :type limit: int, optional

Returns:: items
Return type:: list[dict]

search_data_key(type: FieldType)¶

search_key_factory(model_name, search_key)¶

serch_index(type: FieldType)¶

sk(model_name)¶

sk2model(sk)¶

sk2pk(sk)¶

update(item, batch=None, old_item=None)¶

ddb_single.table.default_pk2model(pk)¶

ddb_single.table.default_pk_factory(model_name)¶

ddb_single.table.default_sk_factory(model_name, prefix='', suffix='_item')¶

ddb_single.table.hash_key(name)¶

ddb_single.table.range_key(name)¶

ddb_single.utils_botos module¶

class ddb_single.utils_botos.FilterStatus(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)¶

Bases: Enum

FILTER = 'filter'¶

SEATCH = 'search'¶

STAGED = 'staged'¶

class ddb_single.utils_botos.QueryType(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)¶

Bases: Enum

Enum for query types

BEGINS = 'BEGINS_WITH'¶

BETWEEN = 'BETWEEN'¶

CONTAINS = 'CONTAINS'¶

EQ = 'EQUAL'¶

EX = 'EXISTS'¶

GT = 'GREATER_THAN'¶

GT_E = 'GREATER_THAN_OR_EQUAL'¶

IN = 'IN'¶

LT = 'LESS_THAN'¶

LT_E = 'LESS_THAN_OR_EQUAL'¶

N_EQ = 'NOT_EQUAL'¶

N_EX = 'NOT_EXISTS'¶

ddb_single.utils_botos.attr_ex(name, value, mode)¶

ddb_single.utils_botos.attr_method(name, value, mode)¶

ddb_single.utils_botos.deserialize(value)¶

ddb_single.utils_botos.is_key(mode)¶

ddb_single.utils_botos.is_same_json(data_0, data_1)¶

ddb_single.utils_botos.json_export(data)¶

ddb_single.utils_botos.json_import(data)¶

ddb_single.utils_botos.range_ex(name, value, mode)¶

ddb_single.utils_botos.serialize(value)¶

ddb_single package¶

Submodules¶

ddb_single.error module¶

ddb_single.model module¶

ddb_single.query module¶

ddb_single.table module¶

ddb_single.utils_botos module¶

Module contents¶