A lightweight and asynchronous ActiveRecord-style wrapper for SQLAlchemy. Brings Django-like queries, automatic timestamps, nested eager loading, and serialization/deserialization for SQLAlchemy models.
Heavily inspired by sqlalchemy-mixins.
Visit the documentation website.
- 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_atandupdated_atfields. - Serialization/deserialization: Serialize and deserialize models to/from dict or JSON easily.
Python>=3.10SQLAlchemy>=2.0.0typing_extensions>=4.5.0 (for Python<3.11)
You can simply install sqlactive from PyPI:
pip install sqlactiveThe ActiveRecordBaseModel class provides a base class for your models.
It inherits from:
ActiveRecordMixin: Provides a set of ActiveRecord-like helper methods for interacting with the database.TimestampMixin: Adds thecreated_atandupdated_attimestamp columns.SerializationMixin: Provides serialization and deserialization methods.
It is recommended to define a BaseModel class that inherits from
ActiveRecordBaseModel and use it as the base class for all models
as shown in the following example:
fromsqlalchemyimportString, ForeignKeyfromsqlalchemy.ormimportMapped, mapped_column, relationshipfromsqlactiveimportActiveRecordBaseModel# Define a base class for your models (recommended)classBaseModel(ActiveRecordBaseModel):
__abstract__=True# Define the modelsclassUser(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')
comments: Mapped[list['Comment']] =relationship(back_populates='user')
classPost(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')
comments: Mapped[list['Comment']] =relationship(back_populates='post')
classComment(BaseModel):
__tablename__='comments'id: Mapped[int] =mapped_column(primary_key=True, autoincrement=True, index=True)
body: Mapped[str] =mapped_column(nullable=False)
post_id: Mapped[int] =mapped_column(ForeignKey('posts.id'))
user_id: Mapped[int] =mapped_column(ForeignKey('users.id'))
post: Mapped['Post'] =relationship(back_populates='comments')
user: Mapped['User'] =relationship(back_populates='comments')Warning
When defining a BaseModel class, don't forget to set __abstract__ to
True in the base class to avoid creating tables for the base class.
Tip
The models can directly inherit from the ActiveRecordBaseModel class:
fromsqlactiveimportActiveRecordBaseModelclassUser(ActiveRecordBaseModel):
__tablename__='users'# ...However, it is recommended to define a base model class for your models and inherit from it.
Your base model class can also inherit directly from the mixins.
For example, if you don't want to implement automatic timestamps don't inherit
from ActiveRecordBaseModel class. Instead, inherit from ActiveRecordMixin
and/or SerializationMixin:
fromsqlactiveimportActiveRecordMixin, SerializationMixinclassBaseModel(ActiveRecordMixin, SerializationMixin):
__abstract__=Truefromasyncioimportcurrent_taskfromsqlalchemy.ext.asyncioimport (
create_async_engine,
async_sessionmaker,
async_scoped_session,
)
fromsqlactiveimportActiveRecordBaseModel# Connect to the databaseDATABASE_URL='sqlite+aiosqlite://'async_engine=create_async_engine(DATABASE_URL, echo=False)
async_sessionmaker=async_sessionmaker(
bind=async_engine,
expire_on_commit=False,
)
async_scoped_session=async_scoped_session(
async_sessionmaker,
scopefunc=current_task,
)
# Set the sessionBaseModel.set_session(async_scoped_session)
# Initialize the tablesasyncwithasync_engine.begin() asconn:
awaitconn.run_sync(BaseModel.metadata.create_all)The use of the expire_on_commit flag is explained in the warning of
this section.
Tip
Use the DBConnection class as a shortcut to initialize the database.
The DBConnection class is a wrapper around the async_engine,
async_sessionmaker and async_scoped_session objects:
fromsqlactiveimportDBConnectionDATABASE_URL='sqlite+aiosqlite://'conn=DBConnection(DATABASE_URL, echo=False)
awaitconn.init_db(BaseModel)Check the documentation of DB Connection Helper for more information.
user=awaitUser.insert(username='John1234', name='John Doe', age=25)
user# <User #1>user.name='Johnny Doe'user['age'] =30awaituser.save()
user.name# Johnny Doeuser=awaitUser.get(1)
user# <User #1>awaituser.update(name='John Doe', age=20)
user['name'] # John Doeuser.age# 20awaituser.delete()Caution
The delete() and remove() methods are not soft deletes methods.
Both of them will permanently delete the row from the database.
So, if you want to keep the row in the database, you can implement
a custom soft delete method, i.e. using save() method to update
the row with a flag indicating if the row is deleted or not
(i.e. a boolean is_deleted column).
Tip
If you need to create a record for a short period of time, you can use the
with statement:
withUser(name='Bob', age=30) asuser:
...The with statement will create the record and delete it at the end of the
block.
Check the Temporary Records documentation for more information.
Also, check the Active Record Mixin API Reference to see all the available methods.
users= [
User(username='John1234', name='John Doe', age=20),
User(username='Jane1234', name='Jane Doe', age=21),
User(username='Bob1234', name='Bob Doe', age=22),
]
awaitUser.insert_all(users)
users=awaitUser.find(username__endswith='Doe').all()
users# [<User #1>, <User #2>]awaitUser.delete_all(users)
users=awaitUser.find(username__endswith='Doe').all()
users# []Tip
Check the Active Record Mixin API Reference to see all the available methods.
Perform simple and complex queries with eager loading:
fromsqlactiveimportJOINED, SUBQUERYuser=awaitUser.where(name='John Doe').first()
user# <User #1>posts=awaitPost.where(rating__in=[2, 3, 4], user___name__like='%Bi%').all()
posts# [<Post #1>, <Post #2>, <Post #3>]posts=awaitPost.sort('-rating', 'user___name').all()
posts# [<Post #3>, <Post #1>, <Post #2>, <Post #4>, ...]comments=awaitComment.join(Comment.user, Comment.post).unique_all()
comments# [<Comment 1>, <Comment 2>, <Comment 3>, <Comment 4>, <Comment 5>, ...]user=awaitUser.with_subquery(User.posts).first()
user# <User #1>user.posts# [<Post #1>, <Post #2>, <Post #3>]schema= {
User.posts: JOINED,
User.comments: (SUBQUERY, {
Comment.post: SELECT_IN
}),
}
user=awaitUser.with_schema(schema).unique_first()
user.comments[0].post.title# Lorem ipsumWarning
All relations used in filtering/sorting/grouping should be explicitly set,
not just being a backref.
See the About Relationships section for more information.
Tip
Check the Active Record Mixin API Reference to see all the available methods.
For more flexibility, the low-level
filter_expr(),
order_expr(),
column_expr()
and eager_expr()
methods can be used.
Example of filter_expr() method usage
Post.filter(*Post.filter_expr(rating__gt=2, body='text')) # orsession.query(Post).filter(*Post.filter_expr(rating__gt=2, body='text'))
It's like filter in SQLALchemy, but also allows magic operators like
rating__gt.
Important
Low-level filter_expr(), order_expr(), column_expr() and
eager_expr() methods are very low-level and does NOT do magic
Django-like joins. Use smart_query() for that:
query=User.smart_query(
criterion=(or_(User.age==30, User.age==32),),
filters={'username__like': '%8'},
sort_columns=(User.username,),
sort_attrs=('-created_at',),
group_columns=(User.username,),
group_attrs=['age'],
schema={
User.posts: JOINED,
User.comments: (SUBQUERY, {
Comment.post: SELECT_IN
}),
},
)Tip
Check the Smart Query Mixin API Reference
for more details about the smart_query() method and the low-level methods.
To perform native SQLAlchemy queries asynchronously,
you can use the execute() method:
fromsqlalchemyimportselect, funcfromsqlactiveimportActiveRecordBaseModel, executeclassBaseModel(ActiveRecordBaseModel):
__abstract__=TrueclassUser(BaseModel):
__tablename__='users'# ...query=select(User.age, func.count(User.id)).group_by(User.age)
result=awaitexecute(query, BaseModel)
# [(20, 1), (22, 4), (25, 12)]See the Native SQLAlchemy queries documentation for more information.
Timestamps (created_at and updated_at) are automatically managed:
user=awaitUser.insert(username='John1234', name='John Doe', age=25)
user.created_at# 2024-12-28 23:00:51user.updated_at# 2024-12-28 23:00:51awaitasyncio.sleep(1)
awaituser.update(name='Johnny Doe')
user.updated_at# 2024-12-28 23:00:52Tip
Check the TimestampMixin
class to know how to customize the timestamps behavior.
Models can be serialized/deserialized to/from dictionaries using
the to_dict() and from_dict() methods:
user=awaitUser.insert(username='John1234', name='John Doe', age=25)
user_dict=user.to_dict()
user_dict# {'id': 1, 'username': 'John1234', 'name': 'John Doe', ...}user=User.from_dict(user_dict)
user.name# John DoeAlso, models can be serialized/deserialized to/from JSON using
the to_json() and from_json() methods:
user=awaitUser.insert(username='John1234', name='John Doe', age=25)
user_json=user.to_json()
user_json# {"id": 1, "username": "John1234", "name": "John Doe", ...}user=User.from_json(user_json)
user.name# John DoeFind the complete documentation here.
Please read the contribution guidelines.
This project is licensed under the MIT License - see the LICENSE file for details.
If you find this project useful, give it a ⭐ on GitHub to show your support!
Also, give it a ⭐ to sqlalchemy-mixins which inspired this project!