SQLite Database

Submodules

Module contents

Database

class sqlite_database.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 database 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.

get_table()[source]

Return table instance

has_many(related: ~typing.Type[~sqlite_database.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 hook(name: str)[source]

Register a hook

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 validator(column_name: str, if_fail: str)[source]

Register a validator

classmethod where(**kwargs)[source]

Basic select operation

class sqlite_database.Column(name: str, type_: str, foreign: bool = False, foreign_ref: str | None = None, primary: bool = False, unique: bool = False, nullable: bool = True, default: Any = None, on_delete: Literal['null', 'cascade', 'no act', 'default', 'restrict'] = 'cascade', on_update: Literal['null', 'cascade', 'no act', 'default', 'restrict'] = 'cascade')[source]

Bases: object

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

property default

Default value

property foreign

Is foreign enabled?

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

property type

Type

property unique

Is unique

class sqlite_database.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_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_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 sqlite_database.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 sqlite_database.Primary(column: str)[source]

Bases: Constraint

Primary constraint

apply(type_: BuilderColumn)[source]

Apply this constraint as primary

class sqlite_database.Row[source]

Bases: dict

Attribute Dictionary

class sqlite_database.Table(parent, table: str, columns: Iterable[Column] | None = None, aggresive_select: bool = False)[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(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None = None, limit: int = 0, order: tuple[tuple[str, Literal['asc', 'desc']], ...] | tuple[str, Literal['asc', 'desc']] | None = None)[source]

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_one(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None = None, order: tuple[tuple[str, Literal['asc', 'desc']], ...] | tuple[str, Literal['asc', 'desc']] | None = None)[source]

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(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None = None, what: tuple[str, ...] | Literal['*'] | str | tuple[str] = '*', page: int = 0, length: int = 10, order: tuple[tuple[str, Literal['asc', 'desc']], ...] | tuple[str, Literal['asc', 'desc']] | None = None, flatten: bool = False)[source]

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.

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(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None = None, what: tuple[str, ...] | Literal['*'] | ParsedFn | str | tuple[str] = '*', limit: int = 0, offset: int = 0, order: tuple[tuple[str, Literal['asc', 'desc']], ...] | tuple[str, Literal['asc', 'desc']] | None = None, flatten: bool = False)[source]

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.

  • squash (bool) – Is it squashed?

Returns:

Selected data

Return type:

Queries

select_one(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None = None, what: tuple[str, ...] | Literal['*'] | str | tuple[str] | ParsedFn = '*', order: tuple[tuple[str, Literal['asc', 'desc']], ...] | tuple[str, Literal['asc', 'desc']] | None = None)[source]

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(where: dict[str, Signature | ParsedFn | SubQuery | Any] | list[tuple[str, Signature | SubQuery | ParsedFn]] | None = None, data: dict[str, Any] | None = None, limit: int = 0, order: tuple[tuple[str, Literal['asc', 'desc']], ...] | tuple[str, Literal['asc', 'desc']] | None = None)[source]

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 sqlite_database.Unique(column: str)[source]

Bases: Constraint

Unique constraint

apply(type_: BuilderColumn)[source]

Apply this constraint to an column

sqlite_database.blob(name: str) BuilderColumn[source]

Create a blob column with name

sqlite_database.integer(name: str) BuilderColumn[source]

Create a integer column with name

sqlite_database.real(name: str) BuilderColumn[source]

Create a real column with name

sqlite_database.text(name: str) BuilderColumn[source]

Create a text column with name