OxenORM
Python ORM based on Rust in backend
Install / Use
/learn @Diman2003/OxenORMREADME
🚀 OxenORM - High-Performance Python ORM Backed by Rust
OxenORM is a revolutionary hybrid Object-Relational Mapper that combines the familiar Pythonic developer experience with the blazing-fast performance of Rust. Built according to RFC 0001, it delivers 10-20× speed-ups versus popular pure-Python ORMs while maintaining full Python compatibility.
🎯 Key Features
⚡ Performance
- 10-20× faster than SQLAlchemy, Tortoise ORM, and Django ORM
- Rust-powered database operations with zero GIL overhead
- Async-first design with deterministic concurrency
- Connection pooling with health checks and exponential backoff
- Query caching with TTL support and performance monitoring
🐍 Pythonic Experience
- Dataclass-style model declarations (Django/Tortoise-like)
- Familiar API - no learning curve for Python developers
- Full type hints support with IDE autocomplete
- Async/await throughout the entire stack
🗄️ Database Support
- PostgreSQL - Full feature support with asyncpg
- MySQL/MariaDB - Complete compatibility
- SQLite - Perfect for development and testing
- Multi-database - Connect to multiple databases simultaneously
🛡️ Safety & Reliability
- Memory safety guaranteed by Rust's type system
- Data race freedom with async/await
- SQL injection protection with parameterized queries
- Compile-time SQL validation (optional)
🛠️ Production Ready
- Comprehensive CLI for database management and migrations
- Production configuration with environment-based settings
- Advanced logging with structured JSON output
- Security features with file upload validation
- Performance monitoring with detailed metrics
- Error handling and validation systems
🔄 Advanced Query System
- Complex queries with field lookups (
age__gt,name__icontains) - Q objects for advanced filtering and logical operations
- Aggregations with support for all major database functions
- Window functions and Common Table Expressions (CTEs)
- Bulk operations with optimized performance
✏️ Robust Update Operations
- Model.update() - Instance-level updates with field validation
- QuerySet.update() - Bulk updates with complex conditions
- Field validation - Type checking and conversion during updates
- Field lookups - Support for comparison operators in updates
- Q objects - Complex conditional updates
🏗️ Architecture
┌─────────────────────────────────────────────────────────────┐
│ Python Layer │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ Models & │ │ QuerySet │ │ Manager │ │
│ │ Fields │ │ API │ │ Interface │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ CLI Tools │ │ Config │ │ Logging │ │
│ │ & Migrations │ │ Management │ │ System │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ PyO3 FFI Bridge │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ Async │ │ Type │ │ Error │ │
│ │ Wrapper │ │ Conversion │ │ Handling │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Rust Core (oxen_engine) │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ SQL Builder │ │ Executor │ │ Connection │ │
│ │ (SQLx AST) │ │ (tokio) │ │ Pool │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ Migration │ │ Serde Layer │ │ Query │ │
│ │ Planner │ │ (PyO3) │ │ Cache │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
│ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────┐ │
│ │ File I/O │ │ Image │ │ Performance│ │
│ │ Operations │ │ Processing │ │ Monitoring │ │
│ └─────────────────┘ └─────────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────┐
│ Database Layer │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ PostgreSQL │ │ MySQL │ │ SQLite │ │
│ │ (asyncpg) │ │ (sqlx) │ │ (sqlx) │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
🚀 Quick Start
Installation
# Basic installation
pip install oxen-orm
# With database drivers
pip install "oxen-orm[postgres,mysql,sqlite]"
# With performance optimizations (uvloop)
pip install "oxen-orm[performance]"
# Development installation
pip install "oxen-orm[dev]"
# Full installation with all features
pip install "oxen-orm[dev,postgres,mysql,sqlite,performance]"
# Or build from source
git clone https://github.com/Diman2003/OxenORM.git
cd OxenORM
pip install -e .
Basic Usage
import asyncio
from oxen import Model, IntegerField, CharField, connect
# Define your models
class User(Model):
id = IntegerField(primary_key=True)
name = CharField(max_length=100)
email = CharField(max_length=255, unique=True)
async def main():
# Connect to database
await connect("postgresql://user:pass@localhost/mydb")
# Create tables
await User.create_table()
# Create records
user = await User.create(name="John Doe", email="john@example.com")
# Query records
users = await User.filter(name__icontains="John")
for user in users:
print(f"Found user: {user.name}")
# Update records
await user.update(name="Jane Doe")
# Delete records
await user.delete()
# Run the async function
asyncio.run(main())
Advanced Features
# Multi-database support
from oxen import MultiDatabaseManager
async def multi_db_example():
manager = MultiDatabaseManager({
'primary': 'postgresql://user:pass@localhost/primary',
'analytics': 'mysql://user:pass@localhost/analytics',
'cache': 'sqlite://:memory:'
})
# Use different databases for different models
await User.objects.using('primary').create(name="User")
await AnalyticsEvent.objects.using('analytics').create(event="page_view")
# Complex queries with advanced features
users = await User.filter(
age__gte=18,
email__icontains="@gmail.com"
).exclude(
is_active=False
).order_by('-created_at').limit(10)
# Advanced field types
class Post(Model):
id = IntegerField(primary_key=True)
title = CharField(max_length=200)
tags = ArrayField(element_type="text") # PostgreSQL array
metadata = JSONBField() # PostgreSQL JSONB
location = GeometryField() # PostgreSQL geometry
file = FileField(upload_to="uploads/") # File handling
image = ImageField(upload_to="images/") # Image processing
# Window functions and CTEs
from oxen.expressions import WindowFunction, CommonTableExpression
# Window function
ranked_users = await User.annotate(
rank=WindowFunction("ROW_NUMBER()", order_by=["created_at DESC"])
).filter(rank__lte=10)
# Common Table Expression
cte = CommonTableExpression("user_stats", User.aggregate(total=Count('id')))
Production Features
# CLI Database Management
# oxen db init --url postgresql://user:pass@localhost/mydb
# oxen db status --url postgresql://user:pass@localhost/mydb
# Migration Management
# oxen migrate makemigrations --url postgresql://user:pass@localhost/mydb
# oxen migrate migrate --url postgresql://user:pass@localhost/mydb
# Performance Benchmarking
# oxen benchmark performance --url postgresql://user:pass@localhost/mydb --iterations 1000
# Interactive Shell
# oxen shell shell --url postgresql://user:pass@localhost/mydb --models myapp.models
# Schema Inspection
# oxen inspect --url postgresql://user:pass@localhost/mydb --output schema.json
📊 Performance Benchmarks
Performance Comparison Charts

Comprehensive performance comparison across all major operations
Speedup Analysis

OxenORM speedup factors vs SQLAlchemy 2.0 across different operations
Feature Comparison
.
qqbot-media
349.2kQQBot 富媒体收发能力。使用 <qqmedia> 标签,系统根据文件扩展名自动识别类型(图片/语音/视频/文件)。
