"""Trigger a reset every day at the same time using the schedule library"""
import schedule
import time
import requests
import threading

def run_continuously():
    cease_continuous_run = threading.Event()

    class ScheduleThread(threading.Thread):
        @classmethod
        def run(cls):
            while not cease_continuous_run.is_set():
                interval = schedule.idle_seconds()
                if interval is None:
                    # no more jobs
                    break
                elif interval > 0:
                    # sleep exactly the right amount of time
                    print(f"Time till Analytics reset: {interval} seconds")
                    time.sleep(interval)
                schedule.run_pending()

    continuous_thread = ScheduleThread()
    continuous_thread.start()
    return cease_continuous_run


def analytics_job():
    """trigger at set time"""
    response = requests.get("https://spitfire-interractions.onrender.com/api/analytics")
    print(response.text)

# at 12:01am everyday
schedule.every().day.at("00:01").do(analytics_job)


stop_run_continuously = run_continuously()
if __name__ == "__main__":
    time.sleep(60)
    # Stop the background thread
    stop_run_continuously.set()
jobs/schedule_analytics.py
hngx-org/spitfire-openapi
from flask_sqlalchemy import SQLAlchemy
from flask_session import Session
from flask_cors import CORS
from flask import Flask, session
from flask_bcrypt import Bcrypt
from openapi.config import App_Config
from jobs.schedule_analytics import stop_run_continuously
import time

# Create an instance of Swagger

db = SQLAlchemy()

bcrypt = Bcrypt()

sess = Session()


def create_app():
    """
    Create a new instance of the app with the given configuration.

    :param config_class: configuration class
    :return: app
    """
    # Initialize Flask-

    app = Flask(__name__)
    app.config["SESSION_SQLALCHEMY"] = db
    app.config.from_object(App_Config)
    if app.config["SQLALCHEMY_DATABASE_URI"]:
        print(f"using db")

    # Initialize CORS
    CORS(app, supports_credentials=True)

    # Initialize SQLAlchemy
    db.init_app(app)

    # Initialize sessions
    sess.init_app(app)

    # Initialize Bcrypt
    bcrypt.init_app(app)

    # Importing the models here so it can create the empty tables.
    # The user table was created by default because it was already in use in
    # the routes
    from openapi.models.user import User
    from openapi.models.analytics import Analytics
    from openapi.models.payments import Payments

    from openapi.auth.routes import auth
    from openapi.errors.handlers import error
    from openapi.routes.interractions import conversation
    from openapi.routes.analytics import analytics
    from openapi.payment.routes import payments

    app.register_blueprint(auth)
    app.register_blueprint(error)
    app.register_blueprint(conversation)
    app.register_blueprint(analytics)
    app.register_blueprint(payments)

    # create db tables from models if not exists
    with app.app_context():
        db.create_all()

    return app

openapi/__init__.py
hngx-org/spitfire-openapi
from flask import Blueprint, request, jsonify, session
from openapi import bcrypt
from openapi.models.user import User
from pydantic import ValidationError
from openapi.schemas import RegisterSchema, LoginSchema
from openapi.errors.handlers import CustomError
from openapi.utils import requires_auth

# url_prefix includes /api/auth before all endpoints in blueprint
auth = Blueprint("auth", __name__, url_prefix="/api/auth")


@auth.route("/register", methods=["POST"])
def register():
    """
    This view function facilitates registration of a new user.

    This endpoint allows users to register by providing their name, email, password, and password confirmation.

    Args:
        None (Uses data from the JSON request body).

    Returns:
        JSON response:
        - If successful: A JSON response with status 201 (Created) and user data.
        - If there's a password confirmation mismatch: A JSON response with status 400 (Bad Request) and an error message.
        - If the email already exists: A JSON response with status 400 (Bad Request) and an error message.
        - If an error occurs during registration: A JSON response with status 400 (Bad Request) and an error message.
    """

    data = request.get_json() if request.get_json() != None else request.form
    try:
        data = RegisterSchema(**data)

        email_exists = User.query.filter_by(email=data.email).one_or_none()
        if email_exists:
            return (
                jsonify({"error": "Forbbiden", "message": "Email already exists!"}),
                403,
            )

        hashed_password = bcrypt.generate_password_hash(data.password).decode("utf-8")

        new_user = User(data.name, data.email, hashed_password)
        new_user.insert()

        session["user"] = {"id": new_user.id}

        return (
            jsonify(
                {
                    "message": "User Created Succesfully",
                    "data": new_user.format(),
                }
            ),
            201,
        )
    except ValidationError as e:
        msg = []
        for err in e.errors():
            field=err["loc"][0]
            error=err["msg"]
            if "regex" in error:
                error="Invalid input format,a-z 0-9 _ only"
            msg.append({
                "field": field,
                "error":error
            })
        return (
            jsonify({"error": "Bad Request", "message": msg}),
            400,
        )
    except Exception as error:
        print(f"{type(error).__name__}: {error}")
        return (
            jsonify(
                {
                    "error": "failed",
                    "message": "Internal Error: User not created",
                }
            ),
            500,
        )


@auth.route("/login", methods=["POST"])
def login_user():
    """
    This function facilitates login of an existing user.

    This endpoint allows users to login by providing their email and password.

    Args:
        None (Uses data from the JSON request body).

    Returns:
        JSON response:
        - If successful:
            A JSON response with status 200 (OK) and user data.
        - If data field is inconsistent:
            A JSON response with status 400 (Bad Request) and an error message.
        - If the user does not exist base on the email credential:
            A JSON response with status 404 (Not Found) and an error message.
        - If there's a password confirmation mismatch:
            A JSON response with status 400 (Bad Request) and an error message.
    """
    data = request.get_json()
    try:
        # validate the data received
        data = LoginSchema(**data)

    except ValidationError as e:
        msg = []
        for err in e.errors():
            msg.append({
                "field": err["loc"][0],
                "error":err["msg"]
            })
        return (
            jsonify({"error": "Bad Request", "message": msg}),
            400,
        )
    user = User.query.filter_by(email=data.email).one_or_none()
    if not user:
        raise CustomError(
            "Resource not Found", 404, "User with this email does not exist"
        )
    if not bcrypt.check_password_hash(user.password, data.password):
        raise CustomError("Unauthorized", 401, "Incorrect password")
    session["user"] = {"id": user.id}
    return (jsonify({"message": "success", "data": user.format()}), 200)


# pylint: disable=broad-exception-caught
@auth.route("/@me")
# lets get the user id of the currently loggedin user using requires_auth wrapper
@requires_auth(session)
def see_sess(user_id):
    """
    get the details of current logged in user
    """
    try:
        user = User.query.filter_by(id=user_id).one_or_none()
        return (
            jsonify({"message": "success", "data": user.format()}),
            200,
        )
    except Exception:
        return (
            jsonify(
                {
                    "error": "Internal server error",
                    "message": "It's not you it's us",
                }
            ),
            500,
        )


@auth.route("/logout")
def logout_user():
    """
    This view function logs out the currently authenticated user.

    This endpoint allows a user to log out, effectively clearing their session and
    ending their authenticated session on the server.

    Returns:
        JSON response:
        - If the user is successfully logged out: A JSON response with status 200 (OK)
          and a success message.

    """
    session.pop("user", None)
    return jsonify({"message": "success"}), 200

openapi/auth/routes.py
hngx-org/spitfire-openapi
"""
summary
"""
import os
import datetime
from dotenv import load_dotenv
import openai


load_dotenv(".env")


# pylint: disable=invalid-name
class App_Config:
    """_summary_"""

    SECRET_KEY = os.environ.get("SECRET_KEY", "test")
    openai.api_key = os.environ.get("OPENAI_API_KEY")

    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "SQLALCHEMY_DATABASE_URI", "sqlite:///test.db"
    )
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_ECHO = False

    SESSION_TYPE = "filesystem" if not os.getenv("PROD", None) else "sqlalchemy"
    SESSION_USE_SIGNER = True
    SESSION_COOKIE_SECURE = False
    SESSION_COOKIE_SAMESITE = "None"
    SESSION_COOKIE_HTTPONLY = False
    PERMANENT_SESSION_LIFETIME = datetime.timedelta(days=30)

openapi/config.py
hngx-org/spitfire-openapi
# pylint: disable=cyclic-import
"""_summary_

    Returns:
        _type_: _description_
    """
from flask import Blueprint, jsonify
from openapi import db

# pylint: disable=unused-argument
# pylint: disable=redefined-outer-name
error = Blueprint("error", __name__)


class CustomError(Exception):
    """Exception class for custom errors"""

    def __init__(self, error, code, message):
        """constructor for custom error class

        Args:
            error (_type_): Error Name
            code (_type_): HTTP error code
            message (_type_): error message
        """
        self.error = error
        self.code = code
        self.message = message


# pylint: disable=broad-exception-caught
@error.teardown_app_request
def clean_up(exc):
    """_summary_

    Args:
        exc (_type_): _description_
    """
    try:
        db.session.remove()
    except Exception:
        pass


@error.app_errorhandler(CustomError)
def custom_error(error):
    """app error handler for custom errors"""
    return (
        jsonify({"error": error.error, "message": error.message}),
        error.code,
    )


@error.app_errorhandler(400)
def bad_request(error):
    """_summary_

    Args:
        error (_type_): _description_

    Returns:
        _type_: _description_
    """
    return jsonify({"error": error.name, "message": error.description}), 400


@error.app_errorhandler(404)
def resource_not_found(error):
    """_summary_

    Args:
        error (_type_): _description_

    Returns:
        _type_: _description_
    """
    return jsonify({"error": error.name, "message": error.description}), 404


@error.app_errorhandler(405)
def method_not_allowed(error):
    """_summary_

    Args:
        error (_type_): _description_

    Returns:
        _type_: _description_
    """
    return (
        jsonify({"error": error.name, "message": error.description}),
        405,
    )


@error.app_errorhandler(422)
def cant_process(error):
    """_summary_

    Args:
        error (_type_): _description_

    Returns:
        _type_: _description_
    """
    return jsonify({"error": error.name, "message": error.description}), 422


# pylint: disable=function-redefined
@error.app_errorhandler(429)
def cant_process(error):
    """_summary_

    Args:
        error (_type_): _description_

    Returns:
        _type_: _description_
    """
    return jsonify({"error": error.name, "message": error.description}), 429


@error.app_errorhandler(500)
def server_error(error):
    """_summary_

    Args:
        error (_type_): _description_

    Returns:
        _type_: _description_
    """
    return jsonify({"error": error.name, "message": "Its not you its us"}), 500

openapi/errors/handlers.py
hngx-org/spitfire-openapi
from openapi import db
from openapi.models.base_model import BaseModel


class Analytics(BaseModel):
    __tablename__ = "analytics"

    openai_requests = db.Column(db.Integer(), default=0, nullable=False)
    subscribers = db.Column(db.Integer(), default=0, nullable=False)

    def __repr__(self):
        return "id: {}, openai_requests: {}, subscribers: {}, created_at: {}, updated_at: {}".format(
            self.id,
            self.openai_requests,
            self.subscribers,
            self.created_at,
            self.updated_at,
        )

    def format(self):
        """Return a dictionary representation of the analytics object"""
        return {
            "id": self.id,
            "openai_requests": self.openai_requests,
            "subscribers": self.subscribers,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

openapi/models/analytics.py
hngx-org/spitfire-openapi
"""Base template for the application models"""
from openapi import db
from uuid import uuid4
from datetime import datetime


def generate_uuid():
    """Generate a unique id using uuid4()"""
    return uuid4().hex


# Create a base model class that will contain common functionality
class BaseModel(db.Model):
    """BaseClass for all models"""

    __abstract__ = True

    id = db.Column(
        db.String(60),
        primary_key=True,
        unique=True,
        nullable=False,
        default=generate_uuid,
    )
    created_at = db.Column(db.DateTime(), default=datetime.now, nullable=False)
    updated_at = db.Column(db.DateTime(), default=datetime.now, nullable=False)

    def insert(self):
        """Insert the current object into the database"""
        db.session.add(self)
        db.session.commit()

    def update(self):
        """Update the current object in the database"""
        self.updated_at = datetime.now()
        db.session.commit()

    def delete(self):
        """Delete the current object from the database"""
        db.session.delete(self)
        db.session.commit()

    def format(self):
        """Format the object's attributes as a dictionary"""
        # This method should be overridden in subclasses
        raise NotImplementedError("Subclasses must implement the 'format' method")

openapi/models/base_model.py
hngx-org/spitfire-openapi
from openapi import db
from openapi.models.base_model import BaseModel


class Payments(BaseModel):
    __tablename__ = "payments"

    user_id = db.Column(db.String(60), db.ForeignKey("users.id"), nullable=False)
    amount = db.Column(db.Numeric(6, 2), nullable=False)

    def __repr__(self):
        return "id: {}, user_id: {}, amount: {}, created_at: {},\
              updated_at: {}".format(
            self.id,
            self.user_id,
            self.amount,
            self.created_at,
            self.updated_at,
        )

    def format(self):
        """Return a dictionary representation of the Payment object"""
        return {
            "id": self.id,
            "user_id": self.user_id,
            "amount": self.amount,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

openapi/models/payments.py
hngx-org/spitfire-openapi
from openapi import db
from openapi.models.base_model import BaseModel


class User(BaseModel):
    __tablename__ = "users"

    email = db.Column(db.String(320), nullable=False, unique=True)
    name = db.Column(db.String(60), nullable=False)
    password = db.Column(db.String(128), nullable=False)
    credits = db.Column(db.Integer(), default=3, nullable=False)

    def __init__(self, name, email, password):
        self.email = email
        self.name = name
        self.password = password

    def __repr__(self):
        return "id: {}, name: {}, email: {}, credit: {}, created_at: {},\
              updated_at: {}".format(
            self.id,
            self.name,
            self.email,
            self.credits,
            self.created_at,
            self.updated_at,
        )

    def format(self):
        """Return a dictionary representation of the User object"""
        return {
            "id": self.id,
            "name": self.name,
            "email": self.email,
            "credits": self.credits,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

openapi/models/user.py
hngx-org/spitfire-openapi
"""Payment route handlers"""
from flask import Blueprint, jsonify, session, request
from openapi.models.user import User
from openapi.models.payments import Payments
from openapi.errors.handlers import CustomError
from openapi.schemas  import CreatePaymentSchema
from openapi.utils import requires_auth, get_current_analytics

# TODO: Register this route when payment checks are in place  
# DONE

payments = Blueprint("payments", __name__, url_prefix="/api/payments")


@payments.route("/", methods=["POST"])
@requires_auth(session)
def create_payment(user_id):
    """create payment"""
    data = request.get_json()
    # TODO: Collect payment details and confirm from client


    data = CreatePaymentSchema(**data)

    user = User.query.filter_by(id=user_id).one_or_none()
    if not user:
        raise CustomError("Resource not found", 404, "User does not exist")
    # adding a default amount for subscription in dollars
    payment = Payments(user_id=user.id, amount="3.00")
    payment.insert()
    daily_analytics = get_current_analytics()
    daily_analytics.subscribers += 1
    daily_analytics.update()
    user.credits += 10
    user.update()
    return (
        jsonify(
            {
                "message": "Payment successful",
                "credits": user.credits,
                "user_id": user.id,
            }
        ),
        200,
    )

openapi/payment/routes.py
hngx-org/spitfire-openapi
from flask import Blueprint
from openapi.models.analytics import Analytics
from openapi.utils import get_current_analytics


analytics = Blueprint("analytics", __name__, url_prefix="/api/analytics")


@analytics.route("/")
def reset_each_day():
    """create a new analytics field each day"""
    current_analytics = get_current_analytics()
    print(current_analytics)
    if not current_analytics:
        new_analytics = Analytics()
        new_analytics.insert()
        return "success"

    return "Today's Analytics already exists"

openapi/routes/analytics.py
hngx-org/spitfire-openapi
"""Openai interactions module"""
from flask import Blueprint, request, session, jsonify
import openai
from openai.error import RateLimitError
from collections import defaultdict
from openapi.utils import get_current_analytics, handle_check_credits
import os

conversation = Blueprint("interraction", __name__, url_prefix="/api/chat")


def generate_chat_completion(message, chat_log) -> str:
    """
    Generates a chat completion using the GPT-3.5-turbo model from OpenAI.

    Args:
        message (str): The user input message for the chat completion.
        chat_logs (List[str]): A list of chat logs containing previous messages.

    Returns:
        str: The content of the generated response as a string.
    """
    messages = [
        {"role": "system", "content": f"{chat_log}"},
        {"role": "user", "content": message},
    ]

    current_analytics = get_current_analytics()
    if current_analytics.openai_requests < int(os.getenv("DAILY_LIMIT")):
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo", messages=messages, temperature=0.5, max_tokens=200
        )
        current_analytics.openai_requests += 1
        current_analytics.update()
        return response["choices"][0]["message"]["content"].strip("\n").strip()
    return "Daily limit reached, please try again tomorrow"


#  completion route that handles user inputs and GPT-4 API interactions.
@conversation.route("/completions", methods=["POST"])
@handle_check_credits(session)
def interractions(user):
    """
    Process user input using the GPT-3.5-turbo API and return the response as a JSON object.

    :param user: The user object containing information about the current user.
    :return: JSON object with the response from the GPT-3.5 model API
    """
    chat_logs = defaultdict(list)
    content_type = request.headers.get("Content-Type")
    if content_type == "application/json":
        req = request.get_json()
        if "user_input" and "history" not in req:
            return (
                jsonify(
                    {
                        "message": "Invalid request! Missing 'user_input' or 'history' key."
                    }
                ),
                400,
            )
        history = req.get("history")
        user_input = req.get("user_input")
    else:
        return jsonify({"message": "Content-Type not supported!"}), 406

    if not isinstance(history, list) and not isinstance(user_input, str):
        return (
            jsonify(
                {
                    "message": "Invalid data type for 'history' or 'user_input' field. Must be a valid array or string."
                }
            ),
            400,
        )
    converse = chat_logs.__getitem__(user.id)
    converse.clear()
    converse.append(history)

    try:
        result = generate_chat_completion(message=user_input, chat_log=history)
        # converse.append(f"AI: {result}")
        user.credits -= 1
        user.update()
        return jsonify({"message": result}), 201
    except RateLimitError:
        return (
            jsonify(
                content="The server is experiencing a high volume of requests. Please try again later."
            ),
            400,
        )
    except Exception as error:
        return (
            jsonify(content="An unexpected error occurred. Please try again later."),
            500,
        )


@conversation.route("/", methods=["POST"])
@handle_check_credits(session)
def string_completion(user):
    """
    Process user input using the GPT-3.5-turbo API and return the response as a JSON object.

    :param user: The user object containing information about the current user.
    :return: JSON object with the response from the GPT-3.5-turbo model  API
    """
    content_type = request.headers.get("Content-Type")
    if content_type == "application/json":
        req = request.get_json()
        if "user_input" not in req:
            return (
                jsonify({"message": "Invalid request! Missing 'user_input' key."}),
                400,
            )
        user_input = req.get("user_input")
    else:
        return jsonify({"message": "Content-Type not supported!"}), 406

    if not isinstance(user_input, str):
        return (
            jsonify(
                {
                    "message": "Invalid data type for  'user_input' field. Must be a valid string."
                }
            ),
            400,
        )
    messages = [
        {
            "role": "system",
            "content": "you are a very helpful and professional assistant",
        },
        {"role": "user", "content": user_input},
    ]

    try:
        result = openai.ChatCompletion.create(
            model="gpt-3.5-turbo", messages=messages, temperature=0.5, max_tokens=200
        )
        response = result["choices"][0]["message"]["content"].strip("\n").strip()
        user.credits -= 1
        user.update()
        return jsonify({"message": response}), 201
    except RateLimitError:
        return (
            jsonify(
                content="The server is experiencing a high volume of requests. Please try again later."
            ),
            400,
        )
    except Exception as error:
        return (
            jsonify(content="An unexpected error occurred. Please try again later."),
            500,
        )


@conversation.route("/cron", methods=["GET"])
def cron():
    """
    Returns a JSON object with the key "hello" and the value "world".

    Example Usage:
    ```python
    GET /cron
    ```
    """
    return {"hello": "world"}

openapi/routes/interractions.py
hngx-org/spitfire-openapi
from pydantic import BaseModel, EmailStr, constr, validator


class RegisterSchema(BaseModel):
    email: EmailStr
    name: constr(regex=r'^[a-zA-Z0-9_]+$',to_lower=True, min_length=2, max_length=64)
    password: constr(min_length=8, max_length=64)
    confirm_password: str

    @validator("email")
    def valid_email_length(cls, email):
        if len(email) > 345:
            raise ValueError("Invalid email length")
        return email
    

    @validator("confirm_password")
    def passwords_are_same(cls, confirm_password,values, **kwargs):
        if "password" in values and confirm_password != values["password"]:
            raise ValueError("Passwords do not match")
        return confirm_password


class LoginSchema(BaseModel):
    email: EmailStr
    password: constr(min_length=8, max_length=64)



class CreatePaymentSchema(BaseModel):
    paymentToken : constr(min_length=4, max_length=21) 
    @validator("paymentToken")
    def validate_token(cls, paymentToken):
        if paymentToken != "examplepaymenttoken":
            raise ValueError("Invalid Payment Token")
        return paymentToken

openapi/schemas.py
hngx-org/spitfire-openapi
"""
Helper functions and decorators
"""
from functools import wraps
from openapi import db
from datetime import date, datetime, timedelta
from openapi.models.user import User
from openapi.models.payments import Payments
from openapi.models.analytics import Analytics
from openapi.errors.handlers import CustomError


def requires_auth(session=None):
    """Auth handler"""
    def is_logged_in_wrapper(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            """
            Ensures a user is logged in or returns error

            Parameterss:
                session(dict):
                    - flask session object

            returns
                id(str):
                    - logged in users id
            """
            user = session.get("user")

            if not user:
                raise CustomError("Unauthorized", 401, "You are not logged in")
            user_id = user.get("id")
            return f(user_id, *args, **kwargs)

        return wrapper

    return is_logged_in_wrapper


def query_one_filtered(table, **kwargs):
    """_summary_

    Args:
        table (_type_): _description_

    Returns:
        _type_: _description_
    """
    return db.session.execute(db.select(table).filter_by(**kwargs)).scalar_one_or_none()


# get all items from table based on filter
# args:table=model_class **kwargs=filters
def query_all_filtered(table, **kwargs):
    """_summary_

    Args:
        table (_type_): _description_

    Returns:
        _type_: _description_
    """
    return db.session.execute(db.select(table).filter_by(**kwargs)).scalars().all()


# get first one item from table no filter
def query_one(table):
    """_summary_

    Args:
        table (_type_): _description_

    Returns:
        _type_: _description_
    """
    return db.session.execute(db.select(table)).scalar_one_or_none()


# get all items on table no filter
def query_all(table):
    """_summary_

    Args:
        table (_type_): _description_

    Returns:
        _type_: _description_
    """
    return db.session.execute(db.select(table)).scalars().all()


def chaaracter_validation(user_input):
    word = user_input.split()
    if len(word) <= 20:
        return user_input
    # reduced_word = " ".join(word[:20])
    raise CustomError("payload too long", 413, "the request body is too long")


def get_current_analytics():
    """Get the current day's analytic entry from the database"""
    return db.session.execute(
        db.select(Analytics).filter(
            db.Cast(Analytics.created_at, db.Date())
            == date.today().strftime("%Y-%m-%d")
        )
    ).scalar_one_or_none()

def handle_check_subscription(user):
    """Confirm subcription status of user"""
    payment = db.session.execute(
        db.Select(Payments).filter_by(user_id=user.get("id"))
                            .order_by(db.desc(Payments.created_at))
    ).scalar_one_or_none()
    # print(f"payment:  {payment}") 
    if payment is None:
        # This means user is still on free trial and hasn't
        # made any payment at all
        return True
    if payment.created_at >= (datetime.now() - timedelta(days=30)):
        # this means user has paid in the last 30 days
        return True
    raise CustomError(
        "Subscription Required", 402, "You do not have an active subscription"
    )



def handle_check_credits(session=None):
    """Check user credit"""
    def chat_completion_wrapper(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            user = session.get("user")
            if not user:
                raise CustomError("Unauthorized", 401, "You are not logged in")
            if handle_check_subscription(user) is True:
                user_id = user.get("id")
                user = User.query.get(user_id)
                if user.credits <= 0:
                    raise CustomError(
                        "Payment Required", 402, "You have exhausted your credits"
                    )

            return f(user, *args, **kwargs)

        return wrapper

    return chat_completion_wrapper

openapi/utils.py
hngx-org/spitfire-openapi
from openapi import create_app

app = create_app()


if __name__ == "__main__":
    app.run(debug=True)

run.py
hngx-org/spitfire-openapi