Runing Fastapi server as MP forked process does not allow me to utilize uvicorn with --reload option

[catchthemonster]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question 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

logger = logging.getLogger(__name__)

class MPFactory(MPMixIn):

                """

                concrete implementation of MP bootstrapping of any service

                """

 

                def run(self, any, profileMe, timeMe):

                                # Whenever possible, logs should be the first thing up

                                try:

                                                any.logData['filename'] = f"{any['service']}.log"

                                                mylogImplementation = MyLogger(any.logData)

                                                mylogImplementation.setup()

                                except Exception as lErr:

                                                logger.error("Error occurred:{}".format(lErr))

 

                                # MP seems to ruin this modules logger, so for now using logging.

                                [logger.info](http://logger.info/)('Starting service for %s', [any.app](http://any.app/))

                                app = None

                                ##register

                                match [any.app](http://any.app/):

                                                case 'ASchedulePortalIO':

                                                                from AScheduler import ApiServer

                                                                app = ApiServer(any, profileMe, timeMe)
...
                                                case _:

                                                                logging.Error("Vertical Async service did not match! What matters is we are not starting anything.")

                                if app != None:

                                                try:

                                                                if any.ASGI:

                                                                                api = getattr(app, any.service)

                                                                                [logging.info](http://logging.info/)('App %s started', [any.app](http://any.app/))

    ##issue!                                                                            uvicorn.run(app=api, host=any.host, port=any.port)

                                                                else:

                                                                                [logger.info](http://logger.info/)("ASGI harness disengaged... Synchronous protocol in works...")

                                                except Exception as uvErr:

                                                                logging.error('Error occurred launching uvicorn: %s', uvErr)

                                else:

                                                raise PropFailure("Vertical Async service did not mach! what matters is we are not starting anything.")

 

 

                def mpStart(self, any, profileMe, timeMe):

                                try:

                                                ##mp.get_context("fork")

                                                proc = mp.get_context('fork').Process(target=self.run, args=(any, profileMe, timeMe), name = any.service, daemon=True)

                                                proc.start()

                                                self.runProcesses[[proc.name](http://proc.name/)] = proc

                                                sleep(3)

                                except Exception as pErr:

                                                logger.error("Error occurred: {}".format(pErr))

                                                raise (PropFailure(pErr))

                                return any.service, proc

 

                def mpStop(self, service, proc):

                                success = False

                                logger.warning("Shutting down {} service".format(service.service))

                                try:

                                                if bool(proc):

                                                                proc.terminate()

                                                                proc.join(0.25)

                                                                sleep(3)

                                                                proc.close()

                                                                success = True

                                except Exception as asyncErr:

                                                logger.error("Error occurred: {}".format(asyncErr))

 

                                return success, proc

Description

logger = logging.getLogger(name)

class MPFactory(MPMixIn):

            """

            concrete implementation of MP bootstrapping of any service

            """



            def run(self, any, profileMe, timeMe):

                            # Whenever possible, logs should be the first thing up

                            try:

                                            any.logData['filename'] = f"{any['service']}.log"

                                            mylogImplementation = MyLogger(any.logData)

                                            mylogImplementation.setup()

                            except Exception as lErr:

                                            logger.error("Error occurred:{}".format(lErr))



                            # MP seems to ruin this modules logger, so for now using logging.

                            [logger.info](http://logger.info/)('Starting service for %s', [any.app](http://any.app/))

                            app = None

                            ##register

                            match [any.app](http://any.app/):

                                            case 'ASchedulePortalIO':

                                                            from AScheduler import ApiServer

                                                            app = ApiServer(any, profileMe, timeMe)

...
case _:

                                                            logging.Error("Vertical Async service did not match! What matters is we are not starting anything.")

                            if app != None:

                                            try:

                                                            if any.ASGI:

                                                                            api = getattr(app, any.service)

                                                                            [logging.info](http://logging.info/)('App %s started', [any.app](http://any.app/))

##issue!                                                                            uvicorn.run(app=api, host=any.host, port=any.port)

                                                            else:

                                                                            [logger.info](http://logger.info/)("ASGI harness disengaged... Synchronous protocol in works...")

                                            except Exception as uvErr:

                                                            logging.error('Error occurred launching uvicorn: %s', uvErr)

                            else:

                                            raise PropFailure("Vertical Async service did not mach! what matters is we are not starting anything.")





            def mpStart(self, any, profileMe, timeMe):

                            try:

                                            ##mp.get_context("fork")

                                            proc = mp.get_context('fork').Process(target=self.run, args=(any, profileMe, timeMe), name = any.service, daemon=True)

                                            proc.start()

                                            self.runProcesses[[proc.name](http://proc.name/)] = proc

                                            sleep(3)

                            except Exception as pErr:

                                            logger.error("Error occurred: {}".format(pErr))

                                            raise (PropFailure(pErr))

                            return any.service, proc



            def mpStop(self, service, proc):

                            success = False

                            logger.warning("Shutting down {} service".format(service.service))

                            try:

                                            if bool(proc):

                                                            proc.terminate()

                                                            proc.join(0.25)

                                                            sleep(3)

                                                            proc.close()

                                                            success = True

                            except Exception as asyncErr:

                                            logger.error("Error occurred: {}".format(asyncErr))



                            return success, proc

""" Uvicorn is asking on reaload=True flag an import of fastapi class via app:"module:class" invocation
That is fine, but then how to pass arguments to fatapi class """
I could not find a way aside from leaving fastapi app without args and then injecting configuration via alternative methods...
Any ideas will help, thanks

Operating System

Linux

Operating System Details

No response

FastAPI Version

0.1.0

Python Version

3.10.0

Additional Context

All I am trying to do is utilize --reload uvicorn functionality w fastapi app bootstrapped as MP forked process...