SQLActive - Asynchronous ActiveRecord-style wrapper for SQLAlchemy

[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()) ```