Here’s a bullet point list to explain my situation:

• I am working with FastAPI
• I have written a FastAPI webserver code which has a number of defined get/post endpoints
• The logic for what this webserver does started to become complex, so I moved all of the business logic into a new class called Webserver
• The file containing the code which is related to defining a FastAPI webserver now simply handles conversions of types and forwards requests to an instance of Webserver

Here is an example:

@app.post('/api/send_order')
def send_order(
    fastapi_order_insert_message: FastAPI_OrderInsertMessage,
    request: Request,
    response: Response,
):
    debug_print_pid() ####################################### <- print some debugging
    timestamp = now()                                       #    information
    ip = request.client.host                                # 
                                                            #
    log.info(f'POST /api/send_order ({ip}, {timestamp})') ###

    try:
        return webserver.send_order(fastapi_order_insert_message) # <- the forwarded
                                                                  #    request

    except Exception as error:
        response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR ### <- some logic to
        return FastAPI_ReturnStatus(                                   #    avoid exposing
            status='error',                                            #    implementation
            message='internal server error',                           #    details to api
        ) ##############################################################    users

If you were wondering, FastAPI_OrderInsertMessage and FastAPI_ReturnStatus are just some pydantic types which I defined.

• The application logic contained inside of Webserver became more complex and started interacting with files
• I realized that I needed a way to call a cleanup function webserver.shutdown() which would flush and close these files
• Webserver.__init__ opens some files, so these need to be closed again

My understanding is that there is only one way to do this, and it is facilitated by the FastAPI “lifespan” concept.

Here’s my implementation of that.

webserver = None

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    global webserver
    webserver = Webserver()
    yield
    webserver.shutdown()

This seems to work. Some logging information shows that an instance of Webserver is created when the application runs, and is also torn down when the application is interrupted with a CTRL + C interrupt signal.

You can see that I have resorted to using a global variable webserver such that the endpoints, for example send_order, can access the object webserver. However, this seems like a bit of a code smell.

Have I used the the right approach here, or did I misunderstand something about how the lifespan concept is intended to be used?

Additional comment: I considered using the dynamic nature of Python to attach a new attribute webserver to app, which is an argument to the lifespan function, and also a global variable. However this potentially risks masking some existing attribute, so again I am unsure if this is advisable.