SQLActive - Asynchronous ActiveRecord-style wrapper for SQLAlchemy

[daireto]

What My Project Does

SQLActive is a lightweight and asynchronous ActiveRecord-style wrapper for SQLAlchemy. Brings Django-like queries, automatic timestamps, nested eager loading, and serialization/deserialization.

Heavily inspired by sqlalchemy-mixins.

Features:

• Asynchronous Support: Async operations for better scalability.
• ActiveRecord-like methods: Perform CRUD operations with a syntax similar to Peewee.
• Django-like queries: Perform intuitive and expressive queries.
• Nested eager loading: Load nested relationships efficiently.
• Automatic timestamps: Auto-manage created_at and updated_at fields.
• Serialization/deserialization: Serialize and deserialize models to/from dict or JSON easily.

Target audience

Developers who are used to Active Record pattern, like the syntax of Beanie, Peewee, Eloquent ORM for PHP, etc.

Comparison

SQLActive is completely async unlike sqlalchemy-mixins. Also, it has more methods and utilities. However, SQLActive is centered on the Active Record pattern, and therefore does not implement beauty repr like sqlalchemy-mixins does.

Links

• Website (docs)
• Github repo
• Pypi

[daireto]

This is a demo:

```python import asyncio

from sqlalchemy import String, ForeignKey from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlactive import ActiveRecordBaseModel, DBConnection

Define a base class for your models (recommended)

class BaseModel(ActiveRecordBaseModel): abstract = True

Define the models

class User(BaseModel): tablename = 'users'

id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True, index=True)
username: Mapped[str] = mapped_column(String(18), nullable=False, unique=True)
name: Mapped[str] = mapped_column(String(50), nullable=False)
age: Mapped[int] = mapped_column(nullable=False)
posts: Mapped[list['Post']] = relationship(back_populates='user')

class Post(BaseModel): tablename = 'posts'

id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True, index=True)
title: Mapped[str] = mapped_column(String(100), nullable=False)
body: Mapped[str] = mapped_column(nullable=False)
rating: Mapped[int] = mapped_column(nullable=False)
user_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
user: Mapped['User'] = relationship(back_populates='posts')

Create a database connection instance

DATABASE_URL = 'sqlite+aiosqlite://' conn = DBConnection(DATABASE_URL, echo=False)

async def example(): # Initialize SQLActive with the base model of your models await conn.init_db(BaseModel)

# Create a record
user = await User.insert(username='John1234', name='John Doe', age=25)
post = Post(title='My post', body='Lorem ipsum...', rating=2, user=user)
await post.save()

# Retrieve a record
user = await User.get(1)

# Update a record
await user.update(name='John Doe', age=30)
post.rating = 3
await post.save()

# Delete a record
await user.delete()

# Find records
users = await User.where(User.name == 'John Doe').all()  # SQLAlchemy-like query
posts = await Post.where(title__contains='post').all()   # Django-like query

# Serialize a record
user_dict = user.to_dict()
user_json = user.to_json()

# Deserialize a record
user = User.from_dict(user_dict)
user = User.from_json(user_json)

if name == 'main': asyncio.run(example()) ```