run_server.py

import argparse
from multiprocessing import Process
import requests
from requests.adapters import HTTPAdapter
from urllib3 import Retry
import uvicorn
import webbrowser

from conf.defaults import server_conf

def parse_args():
    parser = argparse.ArgumentParser(
        description='Start model server with optional arguments',
    )
    parser.add_argument(
        '--host',
        default=server_conf['host'],
        help='bind socket to this host'
    )
    parser.add_argument(
        '--port',
        default=str(server_conf['port']),
        help='bind socket to this port',
    )
    parser.add_argument(
        '--debug',
        action='store_true',
        help='display extra information that is helpful for debugging'
    )
    parser.add_argument(
        '--reload',
        action='store_true',
        help='automatically restart server when changes are noticed, for development purposes'
    )
    return parser.parse_args()


def main(args, app_name='model_server.base.api:app') -> None:

    print('CLI args:\n' + str(args))
    server_process = Process(
        target=uvicorn.run,
        args=(app_name,),
        kwargs={
            'app_dir': '.',
            'host': args.host,
            'port': int(args.port),
            'log_level': 'debug',
            'reload': args.reload,
        },
        daemon=(args.reload is False),
    )
    url = f'http://{args.host}:{int(args.port):04d}/status'
    print(url)
    server_process.start()

    try:
        sesh = requests.Session()
        retries = Retry(
            total=5,
            backoff_factor=0.1,
        )
        sesh.mount('http://', HTTPAdapter(max_retries=retries))
        resp = sesh.get(url)
        assert resp.status_code == 200
    except Exception:
        print('Error starting server')
        server_process.terminate()
        exit()

    webbrowser.open(url, new=1, autoraise=True)

    if args.debug:
        print('Running in debug mode')
    print('Type "STOP" to stop server')
    input_str = ''
    while input_str.upper() != 'STOP':
        input_str = input()

    server_process.terminate()
    print('Finished')


if __name__ == '__main__':
    main(parse_args())