updated package to conform to pylint

This commit is contained in:
Firq 2023-12-21 17:11:47 +01:00
parent 832196e77e
commit 03901e1d12
Signed by: Firq
GPG key ID: 4DE1059A4666E89F
22 changed files with 10 additions and 9 deletions

View file

View file

@ -0,0 +1,25 @@
# pylint: disable=multiple-statements,wrong-import-position,wrong-import-order
from gevent.monkey import patch_all; patch_all()
from gevent.pywsgi import WSGIServer
from .app import Application
from .routes import routes
from .routes.requests import routes_requests
from .config import ServerSettings
APP = Application.get_instance()
app, api = APP.app, APP.api
api.register_blueprint(routes)
api.register_blueprint(routes_requests)
HOSTNAME = ServerSettings.HOSTNAME
PORT = ServerSettings.PORT
if __name__ == '__main__':
http_Server = WSGIServer((HOSTNAME, PORT), app)
try:
print(f"Server available on http://{HOSTNAME}:{PORT}/")
print(f"View docs on http://{HOSTNAME}:{PORT}/swagger")
http_Server.serve_forever()
except KeyboardInterrupt:
print("Keyboard interrupt received, stopping...")

View file

@ -0,0 +1,37 @@
# pylint: disable=too-few-public-methods
from datetime import datetime
from flask import Flask
from flask_smorest import Api
from .config import APISettings
class IsSingletonException(Exception):
pass
class Application:
"""
This is a singleton that can be accessed using get_instance()
It has 2 properties
- app: Used for WSGI servers and such
- api: Used for Blueprints
"""
__instance = None
app = Flask(__name__)
app.config.from_object(APISettings)
api = None
alive_since = None
@staticmethod
def get_instance():
if Application.__instance is None:
Application()
return Application.__instance
def __init__(self):
if Application.__instance is not None:
raise IsSingletonException("This class is a singleton")
Application.__instance = self
self.api = Api(self.app)
self.alive_since = datetime.now()

View file

@ -0,0 +1 @@
from .settings import APISettings, DatabaseSettings, ServerSettings

View file

@ -0,0 +1,35 @@
# pylint: disable=too-few-public-methods
from pathlib import Path
class ServerSettings:
HOSTNAME = "localhost"
PORT = 5000
class DatabaseSettings:
DATABASE_DIRECTORY = Path(__file__).parents[1] / "database" / "storage"
class APISettings:
API_TITLE = "Support Organizer"
API_VERSION = "0.1.0-a1"
OPENAPI_VERSION = "3.1.0"
# openapi.json settings
OPENAPI_URL_PREFIX = "/"
OPENAPI_JSON_PATH = "openapi.json"
# swagger settings
OPENAPI_SWAGGER_UI_PATH = "/swagger"
OPENAPI_SWAGGER_UI_URL = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5.10.0/"
SWAGGER_UI_DOC_EXPANSION = "list"
# redoc settings
OPENAPI_REDOC_PATH = "/redoc"
OPENAPI_REDOC_URL = "https://cdn.jsdelivr.net/npm/redoc@next/bundles/redoc.standalone.js"
# Info settings
API_SPEC_OPTIONS = {
'info': {
'description': 'Support Organizer for FGO'
}
}
FILE_SAVE_DIRECTORY = Path(__file__).parents[1] / ".temp"

View file

@ -0,0 +1 @@
from .database import Database

View file

@ -0,0 +1,31 @@
# pylint: disable=too-few-public-methods
import dictdatabase as DDB
from ..config import DatabaseSettings
class Database:
"""
This is a singleton that can be accessed using get_instance()
"""
class IsSingletonException(Exception):
pass
__instance = None
db = DDB
DatabaseSettings.DATABASE_DIRECTORY.mkdir(exist_ok=True, parents=True)
db.config.storage_directory = DatabaseSettings.DATABASE_DIRECTORY
if not db.at("requests").exists():
db.at("requests").create({})
if not db.at("profiles").exists():
db.at("profiles").create({})
@staticmethod
def get_instance():
if Database.__instance is None:
Database()
return Database.__instance
def __init__(self):
if Database.__instance is not None:
raise self.IsSingletonException("This class is a singleton")
Database.__instance = self

View file

@ -0,0 +1,2 @@
*
!.gitignore

View file

@ -0,0 +1,19 @@
from enum import Enum
import marshmallow as ma
class HealthStatus(Enum):
OK = 0
WARNING = 1
ERROR = 2
CRITICAL = 3
class HealthGet(ma.Schema):
alive_since = ma.fields.String()
alive_for = ma.fields.String()
status = ma.fields.Enum(HealthStatus, type=ma.fields.String)
class ApiVersionGet(ma.Schema):
version = ma.fields.String(example="0.1")
class OpenAPIGet(ma.Schema):
pass

View file

@ -0,0 +1,30 @@
import marshmallow as ma
from .requests import RequestStatus
class BaseField(ma.Schema):
id = ma.fields.Integer(description="Atlas Academy id")
image = ma.fields.Url(description="URL to atlas image")
class ServantField(BaseField):
name = ma.fields.String(description="Name of servant", example="Scáthach")
level = ma.fields.Integer(description="Servant level", example=120)
modifiers = ma.fields.List(ma.fields.String(example="HYPER"))
class CeField(BaseField):
name = ma.fields.String(description="Name of CE", example="Holy Night Supper")
mlb = ma.fields.Boolean(description="If the CE should be MLB or not", example=True)
class RequestFieldContent(ma.Schema):
servant = ma.fields.Nested(ServantField)
ce = ma.fields.Nested(CeField)
class RequestPostData(ma.Schema):
username = ma.fields.String(description="Username of the requesting person", example="Firq")
description = ma.fields.String(description="Short description for what the request is used", example="DB 7T attempts with Scathach")
friendcode = ma.fields.String(description="Friendcode of the requesting person", example="000,000,000")
request = ma.fields.Nested(RequestFieldContent)
class RequestDatabaseEntry(RequestPostData):
uuid = ma.fields.String(description="UUID v4 of the request")
status = ma.fields.Enum(RequestStatus)

View file

@ -0,0 +1,27 @@
from enum import Enum
import marshmallow as ma
class RequestStatus(Enum):
CREATED = 0
ACTIVE = 1
COMPLETED = 64
DELETED = 255
class RequestsAll(ma.Schema):
uuid = ma.fields.String(description="UUID v4", example="3f6dd2e5-8cb1-4ce0-82e4-b68edf4ce37a")
status = ma.fields.String()
class RequestsCreate(ma.Schema):
message = ma.fields.String(example="Request 3f6dd2e5-8cb1-4ce0-82e4-b68edf4ce37a was created")
status = ma.fields.Enum(RequestStatus, type=ma.fields.String)
uuid = ma.fields.String(example="3f6dd2e5-8cb1-4ce0-82e4-b68edf4ce37a")
class UuidSchema(ma.Schema):
uuid = ma.fields.String(description="UUID of the request", example="3f6dd2e5-8cb1-4ce0-82e4-b68edf4ce37a")
class RequestsInteraction(ma.Schema):
status = ma.fields.String()
class RequestsInteractionDelete(ma.Schema):
message = ma.fields.String(example="Request 3f6dd2e5-8cb1-4ce0-82e4-b68edf4ce37a was deleted.")
status = ma.fields.Enum(RequestStatus, type=ma.fields.String)

View file

@ -0,0 +1,6 @@
# pylint: disable=wrong-import-position,cyclic-import
from flask_smorest import Blueprint
routes = Blueprint("interface", "interface", url_prefix="/", description="")
from . import version, openapi, health # avoids circular imports problem

View file

@ -0,0 +1,20 @@
from datetime import datetime
from flask.views import MethodView
from ..app import Application
from ..models.interface import HealthGet, HealthStatus
from . import routes as blp
APP = Application.get_instance()
@blp.route("/health")
class ApiVersion(MethodView):
@blp.doc(summary="Returns the status and alive-time of the server")
@blp.response(200, HealthGet, description="Successful operation")
def get(self):
response = {
"alive_since": datetime.strftime(APP.alive_since, "%d.%m.%Y %H:%M:%S"),
"alive_for": str(datetime.now() - APP.alive_since),
"status": HealthStatus.OK
}
return response

View file

@ -0,0 +1,12 @@
from flask import redirect, url_for
from flask.views import MethodView
from ..models.interface import OpenAPIGet
from . import routes as blp
@blp.route("/openapi")
class ApiVersion(MethodView):
@blp.doc(summary="Get the OpenAPI spec as a JSON.")
@blp.response(200, OpenAPIGet, description="Successful operation")
def get(self):
return redirect(url_for('api-docs.openapi_json'))

View file

@ -0,0 +1,6 @@
# pylint: disable=wrong-import-position,cyclic-import
from flask_smorest import Blueprint
routes_requests = Blueprint("requests", "requests", url_prefix="/requests", description="")
from . import base, create, interact # avoids circular imports problem

View file

@ -0,0 +1,23 @@
from flask.views import MethodView
from ...database import Database
from ...models.requests import RequestStatus
from ...models.requestentry import RequestDatabaseEntry
from . import routes_requests as blp
db = Database.get_instance().db
@blp.route("/all")
class AllRequests(MethodView):
@blp.doc(summary="Returns all requests")
@blp.response(200, RequestDatabaseEntry(many=True), description="Successful operation")
def get(self):
db_data = db.at("requests").read()
response = []
for value in db_data.values():
value.update({
"status": RequestStatus(value["status"])
})
response.append(value)
print(response)
return response

View file

@ -0,0 +1,31 @@
from uuid import uuid4
from flask.views import MethodView
from ...database import Database
from ...models.requests import RequestsCreate, RequestStatus
from ...models.requestentry import RequestPostData
from . import routes_requests as blp
db = Database.get_instance().db
@blp.route("/create")
class CreateRequest(MethodView):
@blp.doc(summary="Create a new request")
@blp.arguments(RequestPostData)
@blp.response(200, RequestsCreate, description="Successful operation")
def post(self, data):
print(data)
uuid = str(uuid4())
with db.at("requests").session() as (session, requests):
data.update({
"status": RequestStatus.ACTIVE,
"uuid": uuid
})
requests[uuid] = data
session.write()
response = {
"status": RequestStatus.CREATED,
"message": f"Request {uuid} was created",
"uuid": uuid
}
return response

View file

@ -0,0 +1,40 @@
from flask_smorest import abort
from flask.views import MethodView
from ...database import Database
from ...models.requests import UuidSchema, RequestsInteractionDelete, RequestStatus
from ...models.requestentry import RequestDatabaseEntry
from . import routes_requests as blp
db = Database.get_instance().db
@blp.route("/<string:uuid>")
class InteractRequest(MethodView):
@blp.doc(summary="Returns a specific request by its UID")
@blp.arguments(UuidSchema, location='path', as_kwargs=True)
@blp.response(200, RequestDatabaseEntry, description="Successful operation")
@blp.alt_response(404)
def get(self, uuid):
response = db.at("requests", key=uuid).read()
if response is None:
abort(404, message=f"Request {uuid} does not exist.")
response.update({
"status": RequestStatus(response["status"])
})
return response
@blp.doc(summary="Deletes a specific request by its UID")
@blp.arguments(UuidSchema, location='path', as_kwargs=True)
@blp.response(200, RequestsInteractionDelete, description="Successful operation")
@blp.alt_response(404)
def delete(self, uuid):
if not db.at("requests", key=uuid).exists():
abort(404, message=f"Request {uuid} does not exist.")
with db.at("requests").session() as (session, requests):
del requests[uuid]
session.write()
response = {
"status": RequestStatus.DELETED,
"message": f"Request {uuid} was deleted."
}
return response

View file

@ -0,0 +1,12 @@
from flask.views import MethodView
from ..config import APISettings
from ..models.interface import ApiVersionGet
from . import routes as blp
@blp.route("/version")
class ApiVersion(MethodView):
@blp.doc(summary="Get the REST interface version identification.")
@blp.response(200, ApiVersionGet, description="Successful operation")
def get(self):
return { "version": APISettings.API_VERSION }