API Documentation (LuminaDB)

Subpackages

luminadb.models package
- Submodules
- Module contents
luminadb.workers package
- Submodules
  - luminadb.workers.connection module
  - luminadb.workers.database module
- Module contents

Submodules

Module contents

Database

class luminadb.BaseModel[source]

Bases: object

Base class for all Models using Model API

classmethod all()[source]: Return all values from the table

classmethod atomic()[source]: Perform operations within a transaction.

belongs_to(related_model: Type[T])[source]: Retrieve the related model that this instance belongs to.

classmethod bulk_create(records: list[dict])[source]: Insert multiple records at once.

classmethod bulk_delete(keys: list[Any], key: str)[source]: Delete multiple records using a primary key.

classmethod bulk_update(records: list[dict], key: str | object = <object object>)[source]: Update multiple records using a primary key or provided key.

classmethod count(**kwargs) → int[source]: Return count of matching records.

classmethod create(**kwargs)[source]: Create data based on kwargs

classmethod create_table(db: Database)[source]: Create table according to annotations and schema from __schema__

delete(_BaseModel__primary=<object object>, /)[source]: Delete current data

classmethod exists(**kwargs) → bool[source]: Check if any record matches the query.

classmethod find(amount: int)[source]: Return models relative to the amount

classmethod find_or_fail(amount: int)[source]: Return models relative to the amount and when returned is equal to 0, throws an error

classmethod first(**kwargs)[source]: Return the first matching record or None if no match is found.

classmethod first_or_fail(**kwargs)[source]: Return the first matching record or raise an error if no match is found.

classmethod get_table()[source]: Return table instance

has_many(related: ~typing.Type[~luminadb.models.T], foreign_key: str | object = <object object>)[source]: Ensure related_model has a Foreign key pointing to self

has_one(related_model: Type[T])[source]: Retrieve the related model where this instance is referenced.

classmethod one(**kwargs)[source]: Return exactly one record. Raises error if multiple results exist.

classmethod query()[source]: Return Query Builder related to this model

raw(query: str, params: list[Any] | tuple[Any, ...] | dict[str, Any])[source]: Raw SQL query

to_dict()[source]: Convert model instance to dictionary.

to_safe_instance() → Self[source]: Wrap instance that complies with __hidden__.

update(_BaseModel__primary: str | object = <object object>, /, **kwargs)[source]: Update current data

classmethod upsert(key: str, **kwargs)[source]: Insert or update a record based on primary key.

classmethod where(**kwargs)[source]: Basic select operation

class luminadb.Column[source]

Bases: object

tip: for foreign_ref, you can split with / to separate table and column name. e.g: user/id

property auto_increment: Auto increment enabled?

property default: Default value

property foreign: Is foreign enabled?

classmethod from_json(json: str)[source]: Convert JSON to a column

property name: Column Name

property nullable: Nullable

property on_delete: Delete setting

property on_update: Update setting

property primary: Is primary or not?

property raw_source: Source / Foreign Reference

property source: Source / Foreign Reference

property source_column: Source column / Foreign reference column

to_json()[source]: Convert column info as JSON

property type: Type

property unique: Is unique

class luminadb.Database(path: str, **kwargs)[source]

Bases: object

Sqlite3 database, this provide basic integration.

Custom flags:: strict : Certain actions are prevented when active, i.e, initializing nonexistent tables forgive: Certain actions are replaced when active, i.e, replacing .create_table to .table

when a table exists

check_table(table: str)[source]: Check if table is exists or not.

close()[source]: Close database

property closed: Is database closed?

commit()[source]: Commit changes to database

create_index(index: Index)[source]: Create an index

create_table(table: str, columns: Iterable[Column] | Iterable[BuilderColumn])[source]

Create table

Parameters:

table (str) – Table name
columns (Iterable[Column]) – Columns for table

Returns:

Newly created table

Return type:

Table

cursor() → WithCursor[source]: Create cursor

delete_index(name: str | Index, exists_ok: bool = False)[source]: Drop an index

delete_table(table: str)[source]

Delete an existing table

Parameters:: table (str) – table name

foreign_pragma(bool_state: Literal['ON', 'OFF', ''] = '')[source]: Enable/disable foreign key pragma

optimize()[source]: Optimize current database

property path: Path to SQL Connection

rename_table(old_table: str, new_table: str) → Table[source]: Rename existing table to a new one.

reset_table(table: str, columns: Iterable[Column] | Iterable[BuilderColumn]) → Table[source]: Reset existing table with new, this rewrote entire table than altering it.

rollback()[source]: Rollback changes

shrink_memory()[source]: Shrink memories from database as much as it can.

property sql: SQL Connection

table(table: str, _Database__columns: Iterable[Column] | None = None)[source]: fetch table

tables() → tuple[Table, ...][source]: Return tuple containing all table except internal tables

vacuum()[source]: Vacuum this database

class luminadb.Foreign(column: str, target: str | Type[Model])[source]

Bases: Constraint

Foreign constraint

apply(type_: BuilderColumn)[source]: Apply this constraint to an column

on_delete(constraint: ConstraintEnum)[source]: On delete constraint

on_update(constraint: ConstraintEnum)[source]: On update constraint

resolve()[source]: Resolve if current target is a Model

property target: Target foreign constraint

class luminadb.Index[source]

Bases: object

build_sql()[source]: Return current information as SQL

columns(*columns: str)[source]: Set which index this index is based on

property index_columns: Return columns target

property index_name: Return index name

property index_target: Return index target

property index_unique: Return unique property

name(name: str)[source]: Set this index name

target(target: str)[source]: Set this index target table

to_json()[source]: Return current information as JSON

unique()[source]: Set whether this index is unique

class luminadb.Primary(column: str, auto: bool = False)[source]

Bases: Constraint

Primary constraint

Accepts optional auto flag to enable auto-increment on integer primary columns when using the BuilderColumn API.

apply(type_: BuilderColumn)[source]: Apply this constraint as primary. If auto was requested, enable auto increment on the builder column as well.

property auto: bool: Auto increment

class luminadb.Row[source]

Bases: Mapping[str, T], Generic[T], dict

Attribute Dictionary

class luminadb.Table(parent: Database, table: str, columns: Iterable[Column] | None = None)[source]

Bases: object

Table. Make sure you remember how the table goes.

add_column(column: Column | BuilderColumn)[source]: Add column to table

property auto_commit: Auto commit state of this instance

columns()[source]: Table columns

commit()[source]: Commit changes

count()[source]: Count how much objects/rows stored in this table

Delete row or rows

Parameters:

where (Condition, optional) – Condition to determine deletion See Signature class about conditional stuff. Defaults to None.
limit (int, optional) – Limit deletion by integer. Defaults to 0.
order (Optional[Orders], optional) – Order of deletion. Defaults to None.

Returns:

Rows affected

Return type:

int

Delete a row

Parameters:

where (Condition, optional) – Conditional to determine deletion.
None. (Defaults to)
order (Optional[Orders], optional) – Order of deletion. Defaults to None.

property deleted: Is table deleted

property force_dirty: Force dirty state, whether .selecting() on dirty/uncommitted data is allowed or not

force_nodelete()[source]: Force “undelete” table. Used if table was mistakenly assigned as deleted.

property in_transaction: Returns True if the table is in an active transaction.

insert(data: dict[str, Any])[source]

Insert data to current table

Parameters:: data (Data) – Data to insert. Make sure it’s compatible with the table.
Returns:: Last rowid
Return type:: int

insert_many(datas: list[dict[str, Any]])[source]: Alias to insert_multiple

insert_multiple(datas: list[dict[str, Any]])[source]

Insert multiple values

Parameters:: datas (Iterable[Data]) – Data to be inserted.

property name: Table name

Paginate select

Parameters:

where (Condition, optional) – Confitions to use. Defaults to None.
what (OnlyColumn, optional) – Select what you want. Default to None.
page (int) – Which page number be returned first
length (int, optional) – Pagination length. Defaults to 10.
order (Optional[Orders], optional) – Order. Defaults to None.
flatten (bool) – Flatten returned data into dict of lists. Defaults to False.

Yields:

Generator[Queries, None, None] – Step-by-step paginated result.

rename_column(old_column: str, new_column: str)[source]: Rename existing column to new column

rollback()[source]: Rollback

Select data in current table. Bare .select() returns all data.

Parameters:

where (Condition, optional) – Conditions to used. Defaults to None.
what – (OnlyColumn, ParsedFn, optional): Select what you want. Default to None.
limit (int, optional) – Limit of select. Defaults to 0.
offset (int, optional) – Offset. Defaults to 0
order (Optional[Orders], optional) – Selection order. Defaults to None.
flatten (bool) – Flatten returned data into dict of lists. Defaults to False.

Returns:

Selected data

Return type:

Queries

Select one data

Parameters:

where (Condition, optional) – Condition to use. Defaults to None.
what – (OnlyColumn, optional): Select what you want. Default to None.
order (Optional[Orders], optional) – Order of selection. Defaults to None.

Returns:

Selected data

Return type:

Any

subquery(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None, columns: tuple[str, ...] | Literal['*'] | str, limit: int = 0) → SubQuery[source]: Push subquery to current .select() of other table

Update rows of current table

Parameters:

data (Data) – New data to update
where (Condition, optional) – Condition dictionary. See Signature about how condition works. Defaults to None.
limit (int, optional) – Limit updates. Defaults to 0.
order (Optional[Orders], optional) – Order of change. Defaults to None.

Returns:

Rows affected

Return type:

int

update_one(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None = None, data: dict[str, Any] | None = None, order: tuple[tuple[str, Literal['asc', 'desc']], ...] | tuple[str, Literal['asc', 'desc']] | None = None) → int[source]: Update 1 data only

class luminadb.Unique(column: str)[source]

Bases: Constraint

Unique constraint

apply(type_: BuilderColumn)[source]: Apply this constraint to an column

luminadb.blob(name: str) → BuilderColumn[source]: Create a blob column with name

luminadb.hook(fn_or_name: Callable[[Model], None]) → staticmethod[[Callable[[Model], None]], None][source]
luminadb.hook(fn_or_name: str): Register a hook

luminadb.integer(name: str) → BuilderColumn[source]: Create a integer column with name

luminadb.model(db: Database, type_checking: bool = False)[source]

Initiate Model API compatible classes. Requires target to be a dataclass, the app automatically injects dataclass if this isn’t a dataclass.

Use type_checking if you want automatic runtime type checker.

luminadb.real(name: str) → BuilderColumn[source]: Create a real column with name

luminadb.text(name: str) → BuilderColumn[source]: Create a text column with name

luminadb.validate(fn_or_column: FuncT) → staticmethod[[FuncT], bool][source]
luminadb.validate(fn_or_column: str, reason: str | None = None): Register a validator