Skip to content

File: Broken/Core/BrokenFastAPI.py

Broken.Core.BrokenFastAPI

Hosts

Source code in Broken/Core/BrokenFastAPI.py 
19
20
21
class Hosts:
    LOOPBACK: str = "127.0.0.1"
    WILDCARD: str = "0.0.0.0"

LOOPBACK 

LOOPBACK: str = '127.0.0.1'

WILDCARD 

WILDCARD: str = '0.0.0.0'

DEFAULT_HOST 

DEFAULT_HOST: str = (
    Hosts.WILDCARD
    if BrokenPlatform.OnUnix
    else Hosts.LOOPBACK
)

DEFAULT_PORT 

DEFAULT_PORT: int = 8000

HostType 

HostType = Annotated[
    str,
    Option(
        "--host",
        "-h",
        help="Target Hostname to run the server on",
    ),
]

PortType 

PortType = Annotated[
    int,
    Option(
        "--port",
        "-p",
        help="Target Port to run the server on",
    ),
]

WorkersType 

WorkersType = Annotated[
    int,
    Option(
        "--workers",
        "-w",
        min=1,
        help="Maximum number of simultaneous renders",
    ),
]

QueueType 

QueueType = Annotated[
    int,
    Option(
        "--queue",
        "-q",
        min=1,
        help="Maximum number of requests until 503 (back-pressure)",
    ),
]

BlockType 

BlockType = Annotated[
    bool,
    Option(
        "--block",
        "-b",
        " /--free",
        " /-f",
        help="Block the current thread until the server stops",
    ),
]

BrokenAPI

Source code in Broken/Core/BrokenFastAPI.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
@define
class BrokenAPI:
    api: FastAPI = Factory(FastAPI)
    """The main FastAPI instance"""

    host: str = DEFAULT_HOST
    """Hostname currently being used by the server"""

    port: int = DEFAULT_PORT
    """Port currently being used by the server"""

    # -------------------------------------------|
    # Utilities

    @property
    def api_url(self) -> str:
        return f"http://{self.host}:{self.port}"

    @property
    def openapi(self) -> dict:
        return self.api.openapi()

    @property
    def openapi_json(self) -> str:
        return json.dumps(self.openapi, ensure_ascii=False)

    # -------------------------------------------|
    # Actions

    def launch(self,
        host: HostType=DEFAULT_HOST,
        port: PortType=DEFAULT_PORT,
        queue: QueueType=20,
        block: BlockType=True,
    ) -> None:
        """Serve an instance of the api"""
        self.host, self.port = (host, port)

        # Proxy async converter
        async def serve():
            await uvicorn.Server(uvicorn.Config(
                host=self.host, port=self.port,
                app=self.api, loop="uvloop",
                limit_concurrency=queue,
            )).serve()

        # Start the server
        BrokenWorker.thread(asyncio.run, serve())

        # Hold main thread
        while bool(block):
            time.sleep(1)

    # -------------------------------------------|
    # Cloud providers

    def runpod(self,
        workers: WorkersType=3,
        queue: QueueType=20,
    ) -> None:
        """Run a serverless instance at runpod.io"""
        import runpod

        # Use the cool features of the local server
        BrokenAPI.launch(**locals(), block=False)

        async def wrapper(config: dict) -> dict:
            response = (await self.render(config["input"]))

            # Convert video to base64 for json transportation
            if any(type in response.media_type for type in ("video", "image")):
                response.body = b64encode(response.body).decode("utf-8")

            return dict(
                status_code=response.status_code,
                media_type=response.media_type,
                headers=response.headers,
                content=response.body,
            )

        # Call the render route directly
        runpod.serverless.start(dict(
            handler=wrapper
        ))

api 

api: FastAPI = Factory(FastAPI)

The main FastAPI instance

host 

host: str = DEFAULT_HOST

Hostname currently being used by the server

port 

port: int = DEFAULT_PORT

Port currently being used by the server

api_url 

api_url: str

openapi 

openapi: dict

openapi_json 

openapi_json: str

launch 

launch(
    host: HostType = DEFAULT_HOST,
    port: PortType = DEFAULT_PORT,
    queue: QueueType = 20,
    block: BlockType = True,
) -> None

Serve an instance of the api

Source code in Broken/Core/BrokenFastAPI.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def launch(self,
    host: HostType=DEFAULT_HOST,
    port: PortType=DEFAULT_PORT,
    queue: QueueType=20,
    block: BlockType=True,
) -> None:
    """Serve an instance of the api"""
    self.host, self.port = (host, port)

    # Proxy async converter
    async def serve():
        await uvicorn.Server(uvicorn.Config(
            host=self.host, port=self.port,
            app=self.api, loop="uvloop",
            limit_concurrency=queue,
        )).serve()

    # Start the server
    BrokenWorker.thread(asyncio.run, serve())

    # Hold main thread
    while bool(block):
        time.sleep(1)

runpod 

runpod(
    workers: WorkersType = 3, queue: QueueType = 20
) -> None

Run a serverless instance at runpod.io

Source code in Broken/Core/BrokenFastAPI.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def runpod(self,
    workers: WorkersType=3,
    queue: QueueType=20,
) -> None:
    """Run a serverless instance at runpod.io"""
    import runpod

    # Use the cool features of the local server
    BrokenAPI.launch(**locals(), block=False)

    async def wrapper(config: dict) -> dict:
        response = (await self.render(config["input"]))

        # Convert video to base64 for json transportation
        if any(type in response.media_type for type in ("video", "image")):
            response.body = b64encode(response.body).decode("utf-8")

        return dict(
            status_code=response.status_code,
            media_type=response.media_type,
            headers=response.headers,
            content=response.body,
        )

    # Call the render route directly
    runpod.serverless.start(dict(
        handler=wrapper
    ))