A new sqlalchemy database scheduler I made 🎉

[farahats9]

Hi everyone,

I wanted to use celery beat with a little bit of control in a project I am working on using FastAPI, so there was no viable libraries available and my aim was something similar to django_celery_beat

So in the end I decided to fork an old library and update it and fix any issues, and that's how my sqlalchemy-celery-beat happened.

Some additions I added to the original code is implementing a clocked schedule model as well as making the relationship a generic foreign key which significantly reduce the periodic tasks table clutter and improves efficiency.

Let me know what you think, all feedback is welcome. Its still alpha but I am actually using it on my production server because I had no other options.

[auvipy]

I might consider it as a future official package of celery. we have sqlalchely backend already so we can just add more feature to it i guess

[evanstjabadi]

This is nice @farahats9 🚀

We also had to fork and maintain a separate fork of: https://github.com/AngelLiang/celery-sqlalchemy-scheduler (archived) to use Celery Beat with SQLAlchemy. Our fork: https://github.com/aruba-uxi/rdbbeat.

@auvipy - it'd really be nice if we could make up an official SQLAlchemy's version of https://github.com/celery/django-celery-beat. Would be happy to help where possible.

[farahats9]

@evanstjabadi your project looks good, you have some features I wanted to add like Alembic migrations.
However, it is lacking some of the original features added in celery-sqlalchemy-scheduler as it only has 1 type of scheduler which is Cron. Which is totally fine if that's all you need but for my use case I need the Interval schedule and Clocked schedule a lot.

[evanstjabadi]

Yeah, I do agree that we limited rdbbeat a bit 🙂
The general sentiment on my side is just having a celery org supported library for sqlalchemy schedules that we can use/contribute towards.

[auvipy]

I will check both/all available and try to make a draft official one soon. will ping for review and etc.

[darkpssngr]

Hi, any updates on this? Are you supporting any of them officially?

[bazylhorsey]

Is this compatible with sqlalchemy 2.0?
celery-sqlalchemy-scheduler is now incompatible with sqlalchemy 2.0 and I'm looking for alternatives

[farahats9]

Yes, it works with 1.4.x and 2.0.x

[seanic1979]

@farahats9 - I would like to use this project with MySQL. How would I create the tables necessary to get it working? If I just follow the example code it is looking for tables like "celery_intervalschedule" etc.

[farahats9]

The command to start celery beat will create the required tables in your database, you pass a user that have permissions for creating tables though.

celery -A tasks beat -S sqlalchemy_celery_beat.schedulers:DatabaseScheduler -l info

# tasks.py

from celery import Celery
celery = Celery('tasks', backend=backend, broker=broker_url)  # initialize celery instance

beat_dburi = 'mysql+mysqldb://user:password@127.0.0.1:3306/databasename'
beat_schema = None  # Defaults to 'celery' or you can add 'schema_name' to prepend it to the table names (default behavior for MySQL) so instead of 'celery_intervalschedule' it will be 'schema_name_intervalschedule'

config = {
    'beat_schedule': ...
    'beat_scheduler': ...
    'beat_max_loop_interval': ...
    'beat_dburi': beat_dburi,
    'beat_schema': beat_schema,
    'beat_sync_every': ...
    'timezone': ...
    'worker_max_tasks_per_child': ...
} 

celery.conf.update(config)

I hope it helps, if you run into any problems don't hesitate to create an issue on the repo.

[seanic1979]

Perfect. I do have other questions but it doesn't appear that issues are enabled on the repo. My next questions is an error I'm getting when I try to update the task schedule using the ORM. For some reason when I try something like this:

    session_manager = SessionManager()
    engine, Session = session_manager.create_session(db_conn_str)
    session = Session()
    stmt = update(PeriodicTask).values(schedule=timedelta(seconds=60)).where(PeriodicTask.name == 'index')
    session.execute(stmt)
    session.commit()

I get this error:

  File ".../lib/python3.9/site-packages/sqlalchemy/orm/bulk_persistence.py", line 392, in _get_orm_crud_kv_pairs
    desc._bulk_update_tuples(v),
AttributeError: 'property' object has no attribute '_bulk_update_tuples'

I'm using:

celery==5.3.1
sqlalchemy-celery-beat==0.7.1
SQLAlchemy==2.0.20

Not sure if this is something caused by using your SessionManager or just something I'm doing wrong.

[farahats9]

You need to create the schedule first, then pass its id and discriminator to the updated task query. Here is how its done:

from sqlalchemy_celery_beat.session import SessionManager, session_cleanup
import sqlalchemy as sa
from sqlalchemy_celery_beat.models import PeriodicTask, PeriodicTaskChanged, IntervalSchedule, Period
from datetime import timedelta
from celery import schedules

session_manager = SessionManager()
beat_dburi = 'sqlite:///test.db'
session = session_manager.session_factory(beat_dburi)

with session_cleanup(session):
    # first create the schedule object in the DB
    schedule = IntervalSchedule(every=60, period=Period.SECONDS)
    session.add(schedule)
    session.flush()
    # OR
    schedule = IntervalSchedule.from_schedule(session, schedules.schedule(timedelta(seconds=60))) # this will get or create if not available in DB
    # then update the task
    stmt = sa.update(PeriodicTask).values(schedule_id=schedule.id, discriminator=schedule.discriminator).where(PeriodicTask.name == 'index')
    session.execute(stmt)
    session.commit()
    # very important, since you did a bulk update you must tell the schedule that something changed
    PeriodicTaskChanged.update_from_session(session)

The field schedule is not an actual column in the DB so it cannot be updated in the way you described.

I also enabled issues in the repo so thanks for letting me know it was disabled.

[seanic1979]

Got the basics working. Thanks for the help!