Skip to content

BaseModel

mydborm.model.BaseModel

Inherit from this to define a database model.

Example

class User(BaseModel): tablename = "users" id = IntField(primary_key=True) username = StrField(max_length=100, nullable=False) email = StrField(max_length=255, nullable=False, unique=True) active = BoolField(default=True)

Create table

User.create_table()

Insert

User.create(username="alice", email="alice@example.com")

Query

users = User.all() user = User.get(id=1) devs = User.filter(active=True)

Update

User.update({"active": False}, id=1)

Delete

User.delete(id=1)

all() classmethod

Return all rows as list of dicts.

belongs_to(related_model, foreign_key=None)

Return the parent record this instance belongs to.

Usage

book = Book.get(id=1) author = book.belongs_to(Author, foreign_key="author_id")

If foreign_key is omitted, defaults to: _id e.g. "author_id"

bulk_create(records) classmethod

Insert multiple rows in a single query. Returns number of inserted rows.

Usage

User.bulk_create([ {"username": "alice", "email": "alice@example.com"}, {"username": "bob", "email": "bob@example.com"}, ])

bulk_delete(ids, key='id') classmethod

Delete multiple rows by a list of key values. Returns number of deleted rows.

Usage

User.bulk_delete([1, 2, 3]) User.bulk_delete(["alice", "bob"], key="username")

bulk_update(records, key='id') classmethod

Update multiple rows. Each record must contain the key field. Returns total number of affected rows.

Usage

User.bulk_update([ {"id": 1, "active": False}, {"id": 2, "active": False}, ])

bulk_upsert(records, conflict_key='id', update_fields=None, create_index=True) classmethod

Insert records or update on conflict — dialect aware.

MySQL → INSERT ... ON DUPLICATE KEY UPDATE YugabyteDB → INSERT ... ON CONFLICT DO UPDATE

Parameters:

Name Type Description Default
records list

list of dicts to insert/update

required
conflict_key str

unique field that determines conflict

'id'
update_fields list

fields to update on conflict

None
create_index bool

auto-create UNIQUE index on conflict_keys

True

Returns:

Type Description
int

number of affected rows

count(**kwargs) classmethod

Count rows, optionally filtered.

create(**kwargs) classmethod

Insert a new row. Returns the new row's primary key.

User.create(username="alice", email="alice@example.com")

create_table(if_not_exists=True) classmethod

Create the database table for this model.

delete(**kwargs) classmethod

Delete rows matching kwargs. Returns number of deleted rows.

User.delete(id=1)

drop_table(if_exists=True) classmethod

Drop the database table for this model.

exists(**kwargs) classmethod

Return True if any row matches kwargs.

filter(**kwargs) classmethod

Return all rows matching kwargs.

User.filter(active=True)

from_dict(data) classmethod

Create a ModelInstance from a plain dict WITHOUT saving to DB.

Usage

user = User.from_dict({"id": 1, "username": "alice"}) print(user.username)

from_json(json_str) classmethod

Create a ModelInstance from a JSON string WITHOUT saving to DB.

Usage

user = User.from_json('{"id": 1, "username": "alice"}')

get(**kwargs) classmethod

Return a single row matching kwargs or None.

User.get(id=1)

has_many(related_model, foreign_key=None)

Return all related records where foreign_key = self.pk.

Usage

author = Author.get(id=1) books = author.has_many(Book, foreign_key="author_id")

If foreign_key is omitted, defaults to: _id e.g. "author_id"

many_to_many(related_model, join_table, source_key=None, target_key=None)

Return all related records via a join table.

Usage

student = Student.get(id=1) courses = student.many_to_many( Course, join_table="student_courses", source_key="student_id", target_key="course_id" )

If source_key / target_key are omitted they default to: _id and _id

query() classmethod

Return a QueryBuilder for this model.

Usage

User.query().where("active", True).order_by("name").all()

schema_info() classmethod

Return model schema information — fields, types, constraints.

Usage

info = User.schema_info() for field, details in info["fields"].items(): print(field, details)

update(data, **where_kwargs) classmethod

Update rows matching where_kwargs with data. Returns number of affected rows.

User.update({"active": False}, id=1)

validate_schema(strict=False) classmethod

Compare model field definitions against the live DB schema.

Parameters:

Name Type Description Default
strict bool

if True raises SchemaError on mismatch

False

Returns:

Type Description
dict

{ "table" : "users", "valid" : True | False, "missing_in_db" : ["phone"], "extra_in_db" : ["old_col"], "matched" : ["id", "username", ...]

dict

}

Usage

result = User.validate_schema() User.validate_schema(strict=True)