API Reference

This section provides a comprehensive reference for all OxenORM APIs.

Core Classes

Model

The base class for all OxenORM models.

Class Definition

class Model:
    """Base class for all OxenORM models."""

    class Meta:
        table_name: str = None
        db_table: str = None
        abstract: bool = False
        constraints: List[str] = []
        indexes: List[str] = []

Meta Options

table_name: Custom table name for the model
db_table: Alias for table_name (deprecated)
abstract: If True, model won't create a table
constraints: List of table-level constraints
indexes: List of table-level indexes

Methods

# Create operations
@classmethod
async def create(cls, **kwargs) -> Model

@classmethod
async def bulk_create(cls, objects_data: List[Dict]) -> List[Model]

# Read operations
@classmethod
async def all(cls, using_db=None) -> List[Model]

@classmethod
async def get(cls, **kwargs) -> Model

@classmethod
async def get_or_none(cls, **kwargs) -> Optional[Model]

@classmethod
async def filter(cls, *args, **kwargs) -> QuerySet

# Update operations
async def update(self, **kwargs) -> None

async def save(self, force_insert=False, force_update=False) -> None

# Delete operations
async def delete(self) -> None

@classmethod
async def bulk_update(cls, objects: List[Model], fields: List[str]) -> None

QuerySet

Represents a database query that can be chained and executed.

Class Definition

class QuerySet:
    """Represents a database query."""

    def __init__(self, model_class, db=None)

Methods

# Filtering
def filter(self, *args, **kwargs) -> QuerySet

def exclude(self, *args, **kwargs) -> QuerySet

# Ordering
def order_by(self, *fields) -> QuerySet

def reverse(self) -> QuerySet

# Limiting
def limit(self, limit: int) -> QuerySet

def offset(self, offset: int) -> QuerySet

# Aggregations
def count(self) -> int

def aggregate(self, *aggregations) -> Dict

def group_by(self, *fields) -> QuerySet

# Window functions
def window(self, *window_functions) -> QuerySet

# CTEs
def with_cte(self, *ctes) -> QuerySet

# Related objects
def select_related(self, *fields) -> QuerySet

def prefetch_related(self, *fields) -> QuerySet

def only(self, *fields) -> QuerySet

def defer(self, *fields) -> QuerySet

# Execution
async def _execute(self) -> List[Model]

async def first(self) -> Optional[Model]

async def last(self) -> Optional[Model]

# Update and delete
async def update(self, **kwargs) -> int

async def delete(self) -> int

# Streaming
def stream(self) -> AsyncIterator[Model]

Q Objects

Represents a database query condition.

Class Definition

class Q:
    """Represents a database query condition."""

    def __init__(self, **kwargs)

    def __and__(self, other) -> Q

    def __or__(self, other) -> Q

    def __invert__(self) -> Q

Usage Examples

from oxen import Q

# Simple condition
Q(name="John")

# Complex condition
Q(age__gte=18) & Q(is_active=True)

# OR condition
Q(age__lt=18) | Q(age__gt=65)

# NOT condition
~Q(is_active=False)

Field Types

Basic Fields

CharField

A field for storing character data.

class CharField(Field):
    def __init__(self, max_length: int = None, **kwargs)

Parameters

max_length: Maximum length of the string
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

TextField

A field for storing large text data.

class TextField(Field):
    def __init__(self, **kwargs)

Parameters

unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

IntegerField

A field for storing integer values.

class IntegerField(Field):
    def __init__(self, **kwargs)

Parameters

unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
auto_increment: If True, auto-increment field
help_text: Help text for the field

FloatField

A field for storing floating-point values.

class FloatField(Field):
    def __init__(self, **kwargs)

Parameters

unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

BooleanField

A field for storing boolean values.

class BooleanField(Field):
    def __init__(self, **kwargs)

Parameters

unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

DateTimeField

A field for storing date and time values.

class DateTimeField(Field):
    def __init__(self, auto_now=False, auto_now_add=False, **kwargs)

Parameters

auto_now: If True, update field on every save
auto_now_add: If True, set field on creation only
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

DateField

A field for storing date values.

class DateField(Field):
    def __init__(self, **kwargs)

Parameters

unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

TimeField

A field for storing time values.

class TimeField(Field):
    def __init__(self, **kwargs)

Parameters

unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

Advanced Fields

UUIDField

A field for storing UUID values.

class UUIDField(Field):
    def __init__(self, primary_key=False, **kwargs)

Parameters

primary_key: If True, use as primary key
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

JSONField

A field for storing JSON data.

class JSONField(Field):
    def __init__(self, **kwargs)

Parameters

unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

EmailField

A field for storing email addresses.

class EmailField(Field):
    def __init__(self, max_length=254, **kwargs)

Parameters

max_length: Maximum length of the email
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

URLField

A field for storing URLs.

class URLField(Field):
    def __init__(self, max_length=200, **kwargs)

Parameters

max_length: Maximum length of the URL
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

FileField

A field for storing file paths.

class FileField(Field):
    def __init__(self, upload_to="", **kwargs)

Parameters

upload_to: Directory to upload files to
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

ImageField

A field for storing image file paths.

class ImageField(FileField):
    def __init__(self, upload_to="", **kwargs)

Parameters

upload_to: Directory to upload images to
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

ArrayField

A field for storing arrays (PostgreSQL only).

class ArrayField(Field):
    def __init__(self, base_field, **kwargs)

Parameters

base_field: Field type for array elements
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

Relational Fields

ForeignKeyField

A field for creating foreign key relationships.

class ForeignKeyField(RelationalField):
    def __init__(self, to, related_name=None, on_delete=None, **kwargs)

Parameters

to: Target model class
related_name: Name for reverse relationship
on_delete: Action on delete (CASCADE, SET_NULL, etc.)
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

OneToOneField

A field for creating one-to-one relationships.

class OneToOneField(RelationalField):
    def __init__(self, to, related_name=None, on_delete=None, **kwargs)

Parameters

to: Target model class
related_name: Name for reverse relationship
on_delete: Action on delete (CASCADE, SET_NULL, etc.)
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

ManyToManyField

A field for creating many-to-many relationships.

class ManyToManyField(RelationalField):
    def __init__(self, to, through=None, related_name=None, **kwargs)

Parameters

to: Target model class
through: Intermediate model for custom relationships
related_name: Name for reverse relationship
unique: If True, field must be unique
null: If True, field can be NULL
default: Default value for the field
db_index: If True, create database index
help_text: Help text for the field

Aggregations

Count

Count the number of objects.

class Count(Aggregation):
    def __init__(self, field="*", distinct=False)

Parameters

field: Field to count (default: "*")
distinct: If True, count distinct values

Sum

Sum the values of a field.

class Sum(Aggregation):
    def __init__(self, field)

Parameters

field: Field to sum

Avg

Calculate the average of a field.

class Avg(Aggregation):
    def __init__(self, field)

Parameters

field: Field to average

Max

Find the maximum value of a field.

class Max(Aggregation):
    def __init__(self, field)

Parameters

field: Field to find maximum of

Min

Find the minimum value of a field.

class Min(Aggregation):
    def __init__(self, field)

Parameters

field: Field to find minimum of

Window Functions

RowNumber

Add row numbers to results.

class RowNumber(WindowFunction):
    def __init__(self):
        super().__init__("ROW_NUMBER")

Rank

Add rank to results.

class Rank(WindowFunction):
    def __init__(self):
        super().__init__("RANK")

DenseRank

Add dense rank to results.

class DenseRank(WindowFunction):
    def __init__(self):
        super().__init__("DENSE_RANK")

Lag

Get value from previous row.

class Lag(WindowFunction):
    def __init__(self, field, offset=1):
        super().__init__("LAG", field, offset)

Parameters

field: Field to lag
offset: Number of rows to lag

Lead

Get value from next row.

class Lead(WindowFunction):
    def __init__(self, field, offset=1):
        super().__init__("LEAD", field, offset)

Parameters

field: Field to lead
offset: Number of rows to lead

Common Table Expressions

CommonTableExpression

Create a common table expression.

class CommonTableExpression:
    def __init__(self, name, query, recursive=False)

Parameters

name: Name of the CTE
query: QuerySet for the CTE
recursive: If True, create recursive CTE

Database Connection

connect

Connect to a database.

async def connect(url: str, **kwargs) -> None

Parameters

url: Database connection URL
**kwargs: Additional connection parameters

Supported URLs

SQLite: sqlite:///path/to/database.db
PostgreSQL: postgresql://user:pass@host:port/database
MySQL: mysql://user:pass@host:port/database

disconnect

Disconnect from the database.

async def disconnect() -> None

set_database_for_models

Set the database connection for all models.

def set_database_for_models(database) -> None

Parameters

database: Database connection object

Transactions

transaction

Create a database transaction.

async def transaction() -> Transaction

Usage

async with transaction() as txn:
    # Database operations
    await txn.commit()

Transaction

Represents a database transaction.

Methods

async def commit(self) -> None
async def rollback(self) -> None

Signals

pre_save

Signal sent before saving a model.

pre_save = Signal()

post_save

Signal sent after saving a model.

post_save = Signal()

pre_delete

Signal sent before deleting a model.

pre_delete = Signal()

post_delete

Signal sent after deleting a model.

post_delete = Signal()

Usage

from oxen import pre_save, post_save

@pre_save.connect
async def handle_pre_save(sender, instance, **kwargs):
    print(f"About to save {instance}")

@post_save.connect
async def handle_post_save(sender, instance, created, **kwargs):
    print(f"Saved {instance}, created: {created}")

Migrations

MigrationEngine

Engine for managing database migrations.

class MigrationEngine:
    def __init__(self, database_url: str)

Methods

async def create_migrations_table(self) -> None
async def get_applied_migrations(self) -> List[str]
async def apply_migration(self, migration_file: str) -> None
async def rollback_migration(self, migration_file: str) -> None

MigrationGenerator

Generator for creating migration files.

class MigrationGenerator:
    def __init__(self, models: List[Type[Model]])

Methods

def generate_migration(self, name: str) -> str
def get_sql_statements(self) -> List[str]

MigrationRunner

Runner for executing migrations.

class MigrationRunner:
    def __init__(self, database_url: str)

Methods

async def run_migrations(self, migration_dir: str) -> None
async def rollback_migrations(self, migration_dir: str, steps: int = 1) -> None
async def get_migration_status(self, migration_dir: str) -> Dict

Performance Monitoring

QueryOptimizer

Optimizer for analyzing and improving query performance.

class QueryOptimizer:
    def __init__(self)

Methods

async def analyze_query(self, queryset: QuerySet) -> QueryPlan
async def get_suggestions(self, queryset: QuerySet) -> List[str]
async def optimize_query(self, queryset: QuerySet) -> QuerySet

MonitoringDashboard

Dashboard for monitoring database performance.

class MonitoringDashboard:
    def __init__(self)

Methods

async def get_metrics(self) -> Dict[str, Any]
async def get_alerts(self) -> List[Alert]
async def add_alert(self, alert: Alert) -> None
async def export_data(self) -> Dict[str, Any]

Admin Interface

AdminInterface

Interface for managing database schemas and models.

class AdminInterface:
    def __init__(self)

Methods

def register_models(self, models: List[Type[Model]]) -> None
def get_schema_summary(self) -> Dict[str, Any]
def get_table_details(self, table_name: str) -> Optional[Dict[str, Any]]
def generate_schema_diagram(self) -> Dict[str, Any]
def export_schema_json(self, filename: str) -> None

Benchmarking

BenchmarkRunner

Runner for performance benchmarking.

class BenchmarkRunner:
    def __init__(self)

Methods

def create_suite(self, name: str, description: str = "") -> BenchmarkSuite
async def run_benchmark(self, test_name: str, operation: Callable,
                       iterations: int = 1000, warmup_iterations: int = 100) -> BenchmarkResult
def generate_report(self, suite_name: str) -> Dict[str, Any]
def save_report(self, suite_name: str, filename: Optional[str] = None) -> Path

Exceptions

OxenError

Base exception for all OxenORM errors.

class OxenError(Exception):
    pass

ConfigurationError

Raised when there's a configuration error.

class ConfigurationError(OxenError):
    pass

ConnectionError

Raised when there's a database connection error.

class ConnectionError(OxenError):
    pass

ValidationError

Raised when field validation fails.

class ValidationError(OxenError):
    pass

IntegrityError

Raised when database integrity constraints are violated.

class IntegrityError(OxenError):
    pass

DoesNotExist

Raised when a requested object doesn't exist.

class DoesNotExist(OxenError):
    pass

MultipleObjectsReturned

Raised when multiple objects are returned when only one was expected.

class MultipleObjectsReturned(OxenError):
    pass

OperationalError

Raised when there's a database operation error.

class OperationalError(OxenError):
    pass

TransactionError

Raised when there's a transaction error.

class TransactionError(OxenError):
    pass

CLI Commands

Database Management

# Initialize database
oxen db init --url <database_url>

# Check database status
oxen db status --url <database_url>

# Create tables
oxen db create-tables --url <database_url>

Migration Management

# Generate migrations
oxen migrate makemigrations --url <database_url>

# Apply migrations
oxen migrate migrate --url <database_url>

# Check migration status
oxen migrate status --url <database_url>

# Rollback migrations
oxen migrate rollback --url <database_url> --steps <number>

Performance Testing

# Run performance benchmarks
oxen benchmark performance --url <database_url> --iterations <number>

# Generate performance report
oxen benchmark report --output <filename>

Admin Interface

# Start admin interface
oxen admin start --host <host> --port <port>

# Open admin interface in browser
oxen admin open --url <admin_url>

FilesExpand file tree

api_reference.rst

Latest commit

History

api_reference.rst

File metadata and controls

API Reference

Core Classes

Model

QuerySet

Q Objects

Field Types

Basic Fields

CharField

TextField

IntegerField

FloatField

BooleanField

DateTimeField

DateField

TimeField

Advanced Fields

UUIDField

JSONField

EmailField

URLField

FileField

ImageField

ArrayField

Relational Fields

ForeignKeyField

OneToOneField

ManyToManyField

Aggregations

Count

Sum

Avg

Max

Min

Window Functions

RowNumber

Rank

DenseRank

Lag

Lead

Common Table Expressions

CommonTableExpression

Database Connection

connect

disconnect

set_database_for_models

Transactions

transaction

Transaction

Signals

pre_save

post_save

pre_delete

post_delete

Migrations

MigrationEngine

MigrationGenerator

MigrationRunner

Performance Monitoring

QueryOptimizer

MonitoringDashboard

Admin Interface

AdminInterface

Benchmarking

BenchmarkRunner

Exceptions

OxenError

ConfigurationError

ConnectionError

ValidationError

IntegrityError

DoesNotExist

MultipleObjectsReturned

OperationalError

TransactionError