any excellent way for starting a backend-process when start fastapi app, like a scheduler?

[w-Bro]

First Check

• I added a very descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

import rpyc
from apscheduler import events as sche_events
from apscheduler.executors.pool import ProcessPoolExecutor
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.background import BackgroundScheduler
from rpyc.utils.server import ThreadedServer

from common import logger
from config import settings
from db.models.sys_schedule_job import ScheduleJobLog
from db.session import create_session
from schemas.request.job_schema import JobLog


class SchedulerService(rpyc.Service):
    def exposed_add_job(self, func, *args, **kwargs):
        return scheduler.add_job(func, *args, **kwargs)

    def exposed_modify_job(self, job_id, jobstore=None, **changes):
        return scheduler.modify_job(job_id, jobstore, **changes)

    def exposed_reschedule_job(self, job_id, jobstore=None, trigger=None, **trigger_args):
        return scheduler.reschedule_job(job_id, jobstore, trigger, **trigger_args)

    def exposed_pause_job(self, job_id, jobstore=None):
        return scheduler.pause_job(job_id, jobstore)

    def exposed_resume_job(self, job_id, jobstore=None):
        return scheduler.resume_job(job_id, jobstore)

    def exposed_remove_job(self, job_id, jobstore=None):
        return scheduler.remove_job(job_id, jobstore)

    def exposed_get_job(self, job_id):
        return scheduler.get_job(job_id)

    def exposed_get_jobs(self, jobstore=None):
        return scheduler.get_jobs(jobstore)

    def exposed_print_jobs(self, jobstore=None):
        return scheduler.print_jobs(jobstore)

    def exposed_pause(self):
        return scheduler.pause()

    def exposed_resume(self):
        return scheduler.resume()

    def exposed_state(self):
        return scheduler.state()

    def exposed_start(self, **kwargs):
        return scheduler.start(**kwargs)

    def exposed_shutdown(self, **kwargs):
        return scheduler.shutdown(**kwargs)

Description

• I have a scheduler process with rpyc, as well as fastapi can control it by http api. But i have to restart 2 processes(apscheduler and fastapi app) every time i change my code.
• Any way can implement that i just start fastapi app and apscheduler would start later? Like subprocess.popen(), but i have to specify the PYTHON PATH.

Operating System

Linux, Windows

Operating System Details

No response

FastAPI Version

0.70.0

Python Version

Python 3.7.10

Additional Context

No response

[rafsaf]

Note sure about your case, but did you try startup and shutdown events?
https://fastapi.tiangolo.com/advanced/events/

[w-Bro]

Note sure about your case, but did you try startup and shutdown events? https://fastapi.tiangolo.com/advanced/events/

thanks for reply, i had known about events, and i tried it recently like this

app_sub_processes = []


@app.on_event("startup")
    async def init_connect():
        logger.info("App startup event")
        import subprocess
        global app_sub_processes
        app_sub_processes.append({
            "name": "scheduler",
            "command": "python main_scheduler_rpc_server.py",
            "pid": None,
            "process": None
        })

        for sub_proc in app_sub_processes:
            proc = subprocess.Popen(sub_proc["command"])
            sub_proc["process"] = proc
            sub_proc["pid"] = proc.pid
            logger.info(f"sub_process[{sub_proc['name']}]-[{proc.pid}] start")

        if settings.SCHEDULER:
            global scheduler, rpc_server_coon
            import rpyc
            try:
                rpc_server_coon = rpyc.connect(settings.SCHEDULER_RPC_SERVER_IP, settings.SCHEDULER_RPC_SERVER_PORT,
                                               config={"allow_all_attrs": True, "allow_pickle": True})
                scheduler = rpc_server_coon.root
            except ConnectionRefusedError as e:
                logger.error("please start rpc server first")

@app.on_event("shutdown")
async def shutdown_connect():
    global app_sub_processes
    for sub_proc in app_sub_processes:
        sub_proc["process"].terminate()
        logger.info(f"sub_process [{sub_proc['name']}]-[{sub_proc['pid']}] closed")
    logger.info(f"App shutdown event")

it seems work in window but i don't test it deepper or in unix.
The problem is that uvicorn will reload when i am debug(honestly, i don't want to close debug mode) and it seems could not trigger the close event and i could not connect rpc server agine beacuse i got a error.

INFO:     Application startup complete.
Traceback (most recent call last):
  File "main_scheduler_rpc_server.py", line 107, in <module>
    server = ThreadedServer(SchedulerService, port=settings.SCHEDULER_RPC_SERVER_PORT, protocol_config=protocol_config)
  File "xxx\site-packages\rpyc\utils\server.py", line 90, in __init__
    self.listener.bind(address)
OSError: [WinError 10048] 通常每个套接字地址(协议/网络地址/端口)只允许使用一次。

in my opinion, i don't think subprocess.Popen is a graceful way and i am here to find some better way.

[rafsaf]

I'm curious why did you choose such a design path. No offence here but looking at a code, it is just not good. You try to start on startup event, from what I understand, completely independent service that is bind to some port.

Please assume you start your uvicorn server or any other in many processes (using workers param), you will get errors, probably "address already in use" or similar. Another situation: you do have many containers, vms etc. with your fastapi instance. For each one you start process with scheduler. Is that desired? Startup/Shutdown may be although a good place to set up/initialize connection as you do it, but - for me - not for starting, you name it, another service that seems to be not related with fastapi app itself at all.

Anyway, good or bad, I suspect this is not subprocess.Popen fault, this issue can be helpful for you #617 and note this answer #617 (comment).

If you want, you may try to change from async to normal functions, see what happens. Using some "sleep" and prints may help you also. But in general, I would run this scheduler in very different process, not in Startup/Shutdown events and best would be - in different container. Hope it helps :D

[w-Bro]

I know it is strange in ecology of python, for those which was maked for lighter application. What i need is building a complete and integrated app. In my situation, scheduler jobs is always modified with the change of business need and just like what i say, if in different processes, i need to restart them both, and for me or the other colleagues, it's easy to forget to restart one. In stop, when fastapi appis stopped, scheduler is no reason to exist, that's why I need to start some processes and close some process when app start or stop.

Anyway, thanks for you reply and remind, i just noticed that start and stop event will run multi-times when multi-processes, I will close this issue and try to find other way to solve this.