CrowdTLS-server/crowdtls/main.py

import asyncio
import sys
from types import FrameType

import uvicorn
import uvloop

from crowdtls.api import app as app_fastapi
from crowdtls.logs import logger
from crowdtls.scheduler import app as app_rocketry


class CrowdTLS(uvicorn.Server):
    def handle_exit(self, sig: int, frame: FrameType) -> None:
        logger.info("Shutting down CrowdTLS")
        return super().handle_exit(sig, frame)


async def start_server():
    logger.info("Starting CrowdTLS")
    server = CrowdTLS(config=uvicorn.Config(app=app_fastapi, workers=1, loop="uvloop"))

    fastapi = asyncio.create_task(server.serve())
    rocket = asyncio.create_task(app_rocketry.serve())

    await asyncio.wait([rocket, fastapi], return_when=asyncio.FIRST_COMPLETED)


def run():
    if sys.version_info >= (3, 11):
        with asyncio.Runner(loop_factory=uvloop.new_event_loop) as runner:
            runner.run(start_server())
    else:
        uvloop.install()
        asyncio.run(start_server())


if __name__ == "__main__":
    run()