Monday, 29 January 2024

Load fastapi as plugin (via REST) via a controlling context manager

import subprocess
import socket
import os

class UvicornAPIRunner:
    def __init__(self, api_folder, app_module="main:app"):
        self.api_folder = api_folder
        self.app_module = app_module
        self.process = None
        self.port = None

    @staticmethod
    def find_free_port():
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(('', 0))
            return s.getsockname()[1]

    def __enter__(self):
        self.original_cwd = os.getcwd()
        os.chdir(self.api_folder)
        self.port = self.find_free_port()

        command = ['uvicorn', self.app_module, f'--port={self.port}']
        self.process = subprocess.Popen(
                command, 
                stdin=subprocess.PIPE, 
                stdout=subprocess.PIPE, 
                stderr=subprocess.STDOUT
        )

        # Wait for the server to start
        try:
            self.process.stdout.readline()
        except Exception as e:
            print(f"Error while waiting for server to start: {e}")

        return self.port

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.process.terminate()
        self.process.wait()
        os.chdir(self.original_cwd)

with UvicornAPIRunner('.', app_module='main:api') as port:
    print(f"API running on port {port}")

No comments:

Post a Comment

Parse Wikipedia dump

""" This module processes Wikipedia dump files by extracting individual articles and parsing them into a structured format, ...