✨ The Nova API infrastructure, simplified and made with focus on self-hosting and ease of use for developers!
import os
import time
import fastapi # For the API

from slowapi import Limiter # For rate limiting (anti-spam)
from fastapi.middleware.cors import CORSMiddleware # For CORS

from dotenv import load_dotenv # For loading environment variables from the .env file

import handler

load_dotenv() # Loads the .env file, which makes the environment variables accessible via os.getenv() or os.environ
app = fastapi.FastAPI()

#? Makes it so that websites can directly access the API using JavaScript which runs on the client side (in the browser).
#? If this was turned off, then custom front-ends like Better ChatGPT, Chatbot UI ChatGPT Next Web etc. would not be able to access the API.
#! If you're using Cloudflare, you might also need to enable CORS there as well!
app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)

#? The actual core of the API. This is where the magic happens.
#? This is in a separate file to make it easier to maintain.
app.include_router(handler.router)

def get_ip_from_request(request: fastapi.Request) -> str:
    """Get the IP address of the incoming request.
    The reason this function is so complicated is because of proxies like Cloudflare that add the IP address of the user to the headers of the request.

    If you're not using any sort of reverse proxy, you can just use

    return request.client.host

    instead.
    """

    #? Some proxies (like Cloudflare) will add the IP address of the user to the headers of the request.
    xff = None
    if request.headers.get('x-forwarded-for'):
        xff, *_ = request.headers['x-forwarded-for'].split(', ')

    #? The IP address of the user can be in one of three places:
    possible_ips = [
        xff,
        request.headers.get('cf-connecting-ip'),
        request.client.host
    ]

    #? The IP address of the user is the first one that is not None.
    detected_ip = next((i for i in possible_ips if i), None)

    #? Whitelist IPs that should not be rate limited
    #* You can whitelist IPs by adding them to the NO_RATELIMIT_IPS environment variable, separated by a space.
    #* You don't even have to add your entire IP address, you can just add the first few blocks, and these will be whitelisted then.
    for whitelisted_ip in os.getenv('NO_RATELIMIT_IPS', '').split():
        if whitelisted_ip in detected_ip:
            custom_key = f'whitelisted-{time.time()}'
            return custom_key

    return detected_ip

#? Rate limiting is used to prevent spam of the API.
#? key_func tells the Limiter how to identify the user. In this case, it uses the IP address of the user.
#* If you're using Cloudflare, you probably might also want to enable rate limiting there as well!
limiter = Limiter(swallow_errors=True, key_func=get_ip_from_request, default_limits=['2/second', '20/minute', '300/hour'])

#? This is the main route of the API. Just shows some basic information about the API.
#? You could also make a simple HTML page and return that instead.
@app.get('/')
async def index():
    return {
        'hi': 'Welcome to Nova-Lite!',
        'github': 'https://github.com/NovaOSS/nova-lite'
    }

if __name__ == '__main__':
    # run testing server
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=4242)

lite/app.py
NovaOSS/nova-lite
import aiohttp
import fastapi

from providers import closed

router = fastapi.APIRouter()

async def client_request(request: dict) -> dict:
    """Make a request to the target API."""

    #? Start the request
    async with aiohttp.ClientSession() as session:
        async with session.request(
            method=request.get('method', 'POST'),
            url=request['url'],
            data=request.get('data'),
            json=request.get('payload'),
            headers=request.get('headers', {}),
            cookies=request.get('cookies'),
            timeout=aiohttp.ClientTimeout(connect=5, total=500.0)
        ) as response:
            #? Raise an exception if the response has an error
            response.raise_for_status()

            #? Send the response back to the client
            async for response_chunk in response.content.iter_any():
                yield response_chunk

@router.route('/v1/{path:path}', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS'])
async def handle(request: fastapi.Request):
    """Handle a request to the proxy."""

    path = request.url.path.replace('/v1/v1', '/v1') # fix double v1

    #? Try to parse the payload as JSON
    try:
        payload = await request.json()
    except:
        payload = {}

    #? Build the request
    request = {
        'path': path,
        'method': request.method,
        'payload': payload,
        'headers': {
            'Content-Type': 'application/json'
        }
    }

    #? Apply the proxy provider by patching the request
    request = await closed.patch_request(request)

    #? Send the request to the target API
    return fastapi.responses.StreamingResponse(client_request(request))

lite/handler.py
NovaOSS/nova-lite
import random
import aiofiles

GPT_3 = [
    'gpt-3.5-turbo',
    'gpt-3.5-turbo-16k',
    'gpt-3.5-turbo-0613',
    'gpt-3.5-turbo-0301',
    'gpt-3.5-turbo-16k-0613',
]

GPT_4 = GPT_3 + [
    'gpt-4',
    'gpt-4-0314',
    'gpt-4-0613',
]

GPT_4_32K = GPT_4 + [
    'gpt-4-32k',
    'gpt-4-32k-0314',
    'gpt-4-32k-0613',
]

async def conversation_to_prompt(conversation: list) -> str:
    text = ''

    for message in conversation:
        text += f'<|{message["role"]}|>: {message["content"]}\n'

    text += '<|assistant|>:'

    return text

async def random_secret_for(name: str) -> str:
    async with aiofiles.open(f'secrets/{name}.txt', encoding='utf8') as f:
        lines = await f.readlines()

    lines = [line.strip() for line in lines if line.strip()]

    try:
        return random.choice(lines)
    except IndexError as exc:
        raise ValueError(f'Keys missing for "{name}" NO_AUTH') from exc

lite/providers/_utils.py
NovaOSS/nova-lite
from . import _utils

ENDPOINT = 'https://api.openai.com'

async def patch_request(request: dict) -> dict:
    key = await _utils.random_secret_for('closed')

    request['url'] = ENDPOINT + request['path']
    request['headers']['Authorization'] = f'Bearer {key}'

    return request

lite/providers/closed.py
NovaOSS/nova-lite
import os
import fastapi
import aiohttp

from dotenv import load_dotenv

load_dotenv()
app = fastapi.FastAPI()

async def client_request(request: dict) -> dict:
    """Make a request to the target API."""

    #? Start the request
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(connect=5, total=500.0)) as session:
        async with session.request(
            method=request.get('method', 'POST'),
            url=request['url'],
            json=request.get('payload'),
            headers=request.get('headers', {})
        ) as response:
            response.raise_for_status()

            async for response_chunk in response.content.iter_any():
                yield response_chunk

@app.route('/v1/{path:path}', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'OPTIONS'])
async def handle(request: fastapi.Request):
    path = request.url.path.replace('/v1/v1', '/v1') # fix double v1

    try: payload = await request.json()
    except: payload = {}

    request = {
        'path': path,
        'method': request.method,
        'payload': payload,
        'headers': {
            'Content-Type': 'application/json',
            'Authorization': 'Bearer ' + os.environ['OPENAI_API_KEY']
        },
        'url': 'https://api.openai.com' + path
    }

    return fastapi.responses.StreamingResponse(client_request(request))

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=4242)

one-file.py
NovaOSS/nova-lite