mirror of https://github.com/stijndcl/didier
Restructure backend
parent
16f8fedd16
commit
6ba104758b
|
|
@ -0,0 +1,6 @@
|
|||
def handler_404(e=None):
|
||||
return {"error": e or "The requested resource could not be found."}, 404
|
||||
|
||||
|
||||
def handler_422(e):
|
||||
return {"error": e}, 422
|
||||
|
|
@ -0,0 +1,3 @@
|
|||
from discord.ext import ipc
|
||||
|
||||
ipc_client = ipc.Client(secret_key="SOME_SECRET_KEY")
|
||||
|
|
@ -0,0 +1 @@
|
|||
from .dm_route import dm_blueprint
|
||||
|
|
@ -0,0 +1,21 @@
|
|||
from backend.ipc_client import ipc_client
|
||||
import json
|
||||
from quart import Blueprint, request, jsonify
|
||||
|
||||
dm_blueprint: Blueprint = Blueprint("dm", __name__, url_prefix="/dm")
|
||||
|
||||
|
||||
@dm_blueprint.route("/", methods=["POST"])
|
||||
async def post_dm():
|
||||
"""
|
||||
Send a DM to the given user
|
||||
"""
|
||||
data = json.loads((await request.body).decode('UTF-8'))
|
||||
|
||||
dm = await ipc_client.request(
|
||||
"send_dm",
|
||||
user=int(data["userid"]),
|
||||
message=data.get("message")
|
||||
)
|
||||
|
||||
return jsonify({"response": dm})
|
||||
|
|
@ -0,0 +1 @@
|
|||
from .ping_route import ping_blueprint
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
from backend.ipc_client import ipc_client
|
||||
from quart import Blueprint, jsonify
|
||||
from time import time
|
||||
|
||||
ping_blueprint: Blueprint = Blueprint("ping", __name__, url_prefix="/ping")
|
||||
|
||||
|
||||
@ping_blueprint.route("/", methods=["GET"])
|
||||
async def get_ping():
|
||||
"""
|
||||
Send a ping request, monitors bot latency and endpoint time
|
||||
"""
|
||||
latency = await ipc_client.request("get_bot_latency")
|
||||
|
||||
return jsonify({"bot_latency": latency, "response_sent": time()})
|
||||
|
|
@ -0,0 +1 @@
|
|||
from .stats_route import stats_blueprint
|
||||
|
|
@ -0,0 +1 @@
|
|||
from .command_stats_route import command_stats_blueprint
|
||||
|
|
@ -0,0 +1,8 @@
|
|||
from quart import Blueprint
|
||||
|
||||
command_stats_blueprint: Blueprint = Blueprint("command_stats", __name__, "/commands")
|
||||
|
||||
|
||||
@command_stats_blueprint.route("/", methods=["GET"])
|
||||
def get_commands():
|
||||
return {}, 200
|
||||
|
|
@ -0,0 +1,14 @@
|
|||
from .command_stats import command_stats_blueprint
|
||||
from quart import Blueprint
|
||||
|
||||
stats_blueprint: Blueprint = Blueprint("stats", __name__, url_prefix="/stats")
|
||||
stats_blueprint.register_blueprint(command_stats_blueprint)
|
||||
|
||||
|
||||
@stats_blueprint.route("/", methods=["GET"])
|
||||
def get_nested_routes():
|
||||
nested_routes = {
|
||||
"commands": "/stats/commands"
|
||||
|
||||
}
|
||||
return {"nested": nested_routes}, 200
|
||||
|
|
@ -1,45 +1,23 @@
|
|||
from discord.ext import ipc
|
||||
from .error_handlers import handler_404, handler_422
|
||||
from .routes.dm import dm_blueprint
|
||||
from .routes.ping import ping_blueprint
|
||||
from .routes.stats import stats_blueprint
|
||||
from functions.database import custom_commands
|
||||
import json
|
||||
from quart import Quart, jsonify, request
|
||||
from quart import Quart, jsonify
|
||||
from quart_cors import cors
|
||||
from time import time
|
||||
|
||||
|
||||
# Initialize app
|
||||
app = Quart(__name__)
|
||||
# TODO allow_origin=re.compile(r"http://localhost:.*")
|
||||
# needs higher Python & Quart version
|
||||
app = cors(app, allow_origin="*")
|
||||
app.config.from_object(__name__)
|
||||
app.url_map.strict_slashes = False
|
||||
|
||||
|
||||
ipc_client = ipc.Client(secret_key="SOME_SECRET_KEY")
|
||||
|
||||
|
||||
@app.route("/ping", methods=["GET"])
|
||||
async def ping():
|
||||
"""
|
||||
Send a ping request, monitors bot latency and endpoint time
|
||||
"""
|
||||
latency = await ipc_client.request("get_bot_latency")
|
||||
|
||||
return jsonify({"bot_latency": latency, "response_sent": time()})
|
||||
|
||||
|
||||
@app.route("/dm", methods=["POST"])
|
||||
async def send_dm():
|
||||
"""
|
||||
Send a DM to the given user
|
||||
"""
|
||||
data = json.loads((await request.body).decode('UTF-8'))
|
||||
|
||||
dm = await ipc_client.request(
|
||||
"send_dm",
|
||||
user=int(data["userid"]),
|
||||
message=data.get("message")
|
||||
)
|
||||
|
||||
return jsonify({"response": dm})
|
||||
# Register blueprints
|
||||
app.register_blueprint(dm_blueprint)
|
||||
app.register_blueprint(ping_blueprint)
|
||||
app.register_blueprint(stats_blueprint)
|
||||
|
||||
|
||||
@app.route("/custom", methods=["GET"])
|
||||
|
|
@ -63,20 +41,16 @@ async def get_custom_command(command_id):
|
|||
command = custom_commands.get_by_id(command_id)
|
||||
|
||||
if command is None:
|
||||
return page_not_found("")
|
||||
return handler_404("")
|
||||
|
||||
return jsonify(command)
|
||||
|
||||
|
||||
@app.errorhandler(404)
|
||||
def page_not_found(e):
|
||||
return jsonify({"error": "No resource could be found matching the given URL."}), 404
|
||||
return handler_404(e)
|
||||
|
||||
|
||||
@app.errorhandler(422)
|
||||
def unprocessable_entity(e):
|
||||
return jsonify({"error": e}), 422
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app.run()
|
||||
return handler_422(e)
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from settings import DB_HOST, DB_NAME, DB_PASSWORD, DB_USERNAME, DB_DIALECT, DB_DRIVER
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.orm import Session, sessionmaker
|
||||
from urllib.parse import quote_plus
|
||||
|
||||
# Encode password
|
||||
|
|
@ -11,6 +11,6 @@ engine = create_engine(
|
|||
# Format: dialect+driver://username:password@host:port/database
|
||||
f"{DB_DIALECT}{'+' if DB_DRIVER else ''}{DB_DRIVER}://{DB_USERNAME}:{_encoded_pw}@{DB_HOST}/{DB_NAME}"
|
||||
)
|
||||
session = sessionmaker(bind=engine)()
|
||||
session: Session = sessionmaker(bind=engine)()
|
||||
|
||||
Base = declarative_base()
|
||||
|
|
|
|||
|
|
@ -1,5 +1,8 @@
|
|||
from enum import IntEnum
|
||||
from typing import Dict, List
|
||||
|
||||
from database.db import session
|
||||
from database.models import CommandStats
|
||||
from functions.database import utils
|
||||
from functions.stringFormatters import leading_zero as lz
|
||||
import time
|
||||
|
|
@ -75,3 +78,17 @@ def _get_all():
|
|||
|
||||
cursor.execute("SELECT * FROM command_stats")
|
||||
return cursor.fetchall()
|
||||
|
||||
|
||||
def query_command_stats() -> List[Dict]:
|
||||
stats = []
|
||||
|
||||
for instance in session.query(CommandStats).order_by(CommandStats.day):
|
||||
stats.append({
|
||||
"day": instance.day,
|
||||
"commands": instance.commands,
|
||||
"slash_commands": instance.slash_commands,
|
||||
"context_menus": instance.context_menus
|
||||
})
|
||||
|
||||
return stats
|
||||
|
|
|
|||
Loading…
Reference in New Issue