Task can run with thread, but not with process

[ardevik]

worker.main.py

import logging

import asyncio

from apscheduler import AsyncScheduler
from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
from apscheduler.eventbrokers.asyncpg import AsyncpgEventBroker
from apscheduler.executors.subprocess import ProcessPoolJobExecutor
from sqlalchemy.ext.asyncio import create_async_engine
from module.tick.tick import tick

logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)
engine = create_async_engine("postgresql+asyncpg://postgres:secret@localhost/postgres")
data_store = SQLAlchemyDataStore(engine)
event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)
job_executors = {
    "custom_process": ProcessPoolJobExecutor(max_workers=2),
    "custom_thread": ThreadPoolJobExecutor(max_workers=2),
}
scheduler = AsyncScheduler(data_store, event_broker,job_executors=job_executors)

async def main():
    async with scheduler as sch:
        await sch.configure_task("tock", func=tick, max_running_jobs=1)
        await sch.run_until_stopped()

asyncio.run(main())

worker.module.tick.tick.py

from datetime import datetime
from sqlalchemy import create_engine, text


def tick():
    print("Hello, the time is", datetime.now())

    engine = create_engine("postgresql+psycopg://postgres:secret@localhost/postgres")

    with engine.connect() as conn:
        conn.execute(text("INSERT INTO test values(:x)"), {"x": datetime.now()})
        conn.commit()

worker console result

INFO:apscheduler._schedulers.async_:Connection established
DEBUG:apscheduler._schedulers.async_:Started internal cleanup loop with interval: 0:15:00
INFO:apscheduler._schedulers.async_:Scheduler started
INFO:apscheduler._schedulers.async_:Cleaned up expired job results and finished schedules
DEBUG:apscheduler._schedulers.async_:Waiting for any due schedules to appear

It seems like the task function not get fired with process

scheduler.main.py

import logging

import asyncio

from apscheduler import AsyncScheduler
from apscheduler.datastores.sqlalchemy import SQLAlchemyDataStore
from apscheduler.eventbrokers.asyncpg import AsyncpgEventBroker
from apscheduler.triggers.interval import IntervalTrigger
from sqlalchemy.ext.asyncio import create_async_engine

logging.basicConfig(level=logging.DEBUG)
engine = create_async_engine("postgresql+asyncpg://postgres:secret@localhost/postgres")
data_store = SQLAlchemyDataStore(engine)
event_broker = AsyncpgEventBroker.from_async_sqla_engine(engine)
scheduler = AsyncScheduler(data_store, event_broker)

async def main():
    async with scheduler as sch:
        await sch.add_job(func_or_task_id="tock", job_executor="custom_process")

asyncio.run(main())

scheduler console result

DEBUG:asyncio:Using proactor: IocpProactor
INFO:apscheduler._schedulers.async_:Connection established
INFO:apscheduler._schedulers.async_:Stream finished

above code work if i change the job_executor to thread
await sch.add_job(func_or_task_id="tock", job_executor="custom_thread")

INFO:apscheduler._schedulers.async_:Connection established
DEBUG:apscheduler._schedulers.async_:Started internal cleanup loop with interval: 0:15:00
INFO:apscheduler._schedulers.async_:Scheduler started
INFO:apscheduler._schedulers.async_:Cleaned up expired job results and finished schedules
DEBUG:apscheduler._schedulers.async_:Waiting for any due schedules to appear
Hello, the time is 2024-06-30 10:17:19.458513
INFO:apscheduler._schedulers.async_:Job 31c94aa8-bb8f-412b-9aa2-4e4347ec6561 completed successfully

and data get inserted to database

Please help, i need run the function in process

[ardevik]

i found the answer in anyio documentation, we need to check the name before calling asyncio.run(main())

if __name__ == '__main__':
    asyncio.run(main())

[agronholm]

This is the norm when you have a module that can be run directly or imported from elsewhere.