aljaz/CTFd
1
0
Fork 0
mirror of https://github.com/aljazceru/CTFd.git synced 2025-12-17 05:54:19 +01:00
Code Issues Packages Projects Releases Wiki Activity

API Documentation structure (#1480)

Browse Source 
Implement a basic working idea for how to start adding better documentation on the REST API
* Works on #821
This commit is contained in:
Kevin Chung
2020-06-24 10:59:17 -04:00
committed by GitHub
parent 412692d49a
commit a61ff68458
20 changed files with 1074 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
CTFd/api/__init__.py
View File

@@ -9,6 +9,11 @@ from CTFd.api.v1.flags import flags_namespace
from CTFd.api.v1.hints import hints_namespace
from CTFd.api.v1.notifications import notifications_namespace
from CTFd.api.v1.pages import pages_namespace
from CTFd.api.v1.schemas import (
APIDetailedSuccessResponse,
APISimpleErrorResponse,
APISimpleSuccessResponse,
)
from CTFd.api.v1.scoreboard import scoreboard_namespace
from CTFd.api.v1.statistics import statistics_namespace
from CTFd.api.v1.submissions import submissions_namespace
@@ -21,6 +26,12 @@ from CTFd.api.v1.users import users_namespace
api = Blueprint("api", __name__, url_prefix="/api/v1")
CTFd_API_v1 = Api(api, version="v1", doc=current_app.config.get("SWAGGER_UI"))
CTFd_API_v1.schema_model("APISimpleErrorResponse", APISimpleErrorResponse.schema())
CTFd_API_v1.schema_model(
"APIDetailedSuccessResponse", APIDetailedSuccessResponse.schema()
)
CTFd_API_v1.schema_model("APISimpleSuccessResponse", APISimpleSuccessResponse.schema())
CTFd_API_v1.add_namespace(challenges_namespace, "/challenges")
CTFd_API_v1.add_namespace(tags_namespace, "/tags")
CTFd_API_v1.add_namespace(awards_namespace, "/awards")

47
CTFd/api/v1/awards.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_standings
from CTFd.models import Awards, Users, db
from CTFd.schemas.awards import AwardSchema
@@ -9,10 +13,39 @@ from CTFd.utils.decorators import admins_only
awards_namespace = Namespace("awards", description="Endpoint to retrieve Awards")
AwardModel = sqlalchemy_to_pydantic(Awards)
class AwardDetailedSuccessResponse(APIDetailedSuccessResponse):
data: AwardModel
class AwardListSuccessResponse(APIListSuccessResponse):
data: List[AwardModel]
awards_namespace.schema_model(
"AwardDetailedSuccessResponse", AwardDetailedSuccessResponse.apidoc()
)
awards_namespace.schema_model(
"AwardListSuccessResponse", AwardListSuccessResponse.apidoc()
)
@awards_namespace.route("")
class AwardList(Resource):
@admins_only
@awards_namespace.doc(
description="Endpoint to create an Award object",
responses={
200: ("Success", "AwardListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
@@ -57,6 +90,16 @@ class AwardList(Resource):
@awards_namespace.param("award_id", "An Award ID")
class Award(Resource):
@admins_only
@awards_namespace.doc(
description="Endpoint to get a specific Award object",
responses={
200: ("Success", "AwardDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, award_id):
award = Awards.query.filter_by(id=award_id).first_or_404()
response = AwardSchema().dump(award)
@@ -66,6 +109,10 @@ class Award(Resource):
return {"success": True, "data": response.data}
@admins_only
@awards_namespace.doc(
description="Endpoint to delete an Award object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, award_id):
award = Awards.query.filter_by(id=award_id).first_or_404()
db.session.delete(award)

73
CTFd/api/v1/challenges.py
View File

@@ -1,9 +1,12 @@
import datetime
from typing import List
from flask import abort, render_template, request, url_for
from flask_restx import Namespace, Resource
from sqlalchemy.sql import and_
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_standings
from CTFd.models import ChallengeFiles as ChallengeFilesModel
from CTFd.models import (
@@ -47,12 +50,42 @@ challenges_namespace = Namespace(
"challenges", description="Endpoint to retrieve Challenges"
)
ChallengeModel = sqlalchemy_to_pydantic(Challenges)
TransientChallengeModel = sqlalchemy_to_pydantic(Challenges, exclude=["id"])
class ChallengeDetailedSuccessResponse(APIDetailedSuccessResponse):
data: ChallengeModel
class ChallengeListSuccessResponse(APIListSuccessResponse):
data: List[ChallengeModel]
challenges_namespace.schema_model(
"ChallengeDetailedSuccessResponse", ChallengeDetailedSuccessResponse.apidoc()
)
challenges_namespace.schema_model(
"ChallengeListSuccessResponse", ChallengeListSuccessResponse.apidoc()
)
@challenges_namespace.route("")
class ChallengeList(Resource):
@check_challenge_visibility
@during_ctf_time_only
@require_verified_emails
@challenges_namespace.doc(
description="Endpoint to get Challenge objects in bulk",
responses={
200: ("Success", "ChallengeListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
# This can return None (unauth) if visibility is set to public
user = get_current_user()
@@ -132,6 +165,16 @@ class ChallengeList(Resource):
return {"success": True, "data": response}
@admins_only
@challenges_namespace.doc(
description="Endpoint to create a Challenge object",
responses={
200: ("Success", "ChallengeDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
data = request.form or request.get_json()
challenge_type = data["type"]
@@ -162,11 +205,20 @@ class ChallengeTypes(Resource):
@challenges_namespace.route("/<challenge_id>")
@challenges_namespace.param("challenge_id", "A Challenge ID")
class Challenge(Resource):
@check_challenge_visibility
@during_ctf_time_only
@require_verified_emails
@challenges_namespace.doc(
description="Endpoint to get a specific Challenge object",
responses={
200: ("Success", "ChallengeDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, challenge_id):
if is_admin():
chal = Challenges.query.filter(Challenges.id == challenge_id).first_or_404()
@@ -311,6 +363,16 @@ class Challenge(Resource):
return {"success": True, "data": response}
@admins_only
@challenges_namespace.doc(
description="Endpoint to edit a specific Challenge object",
responses={
200: ("Success", "ChallengeDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self, challenge_id):
challenge = Challenges.query.filter_by(id=challenge_id).first_or_404()
challenge_class = get_chal_class(challenge.type)
@@ -319,6 +381,10 @@ class Challenge(Resource):
return {"success": True, "data": response}
@admins_only
@challenges_namespace.doc(
description="Endpoint to delete a specific Challenge object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, challenge_id):
challenge = Challenges.query.filter_by(id=challenge_id).first_or_404()
chal_class = get_chal_class(challenge.type)
@@ -529,7 +595,6 @@ class ChallengeAttempt(Resource):
@challenges_namespace.route("/<challenge_id>/solves")
@challenges_namespace.param("id", "A Challenge ID")
class ChallengeSolves(Resource):
@check_challenge_visibility
@check_score_visibility
@@ -577,7 +642,6 @@ class ChallengeSolves(Resource):
@challenges_namespace.route("/<challenge_id>/files")
@challenges_namespace.param("id", "A Challenge ID")
class ChallengeFiles(Resource):
@admins_only
def get(self, challenge_id):
@@ -593,7 +657,6 @@ class ChallengeFiles(Resource):
@challenges_namespace.route("/<challenge_id>/tags")
@challenges_namespace.param("id", "A Challenge ID")
class ChallengeTags(Resource):
@admins_only
def get(self, challenge_id):
@@ -609,7 +672,6 @@ class ChallengeTags(Resource):
@challenges_namespace.route("/<challenge_id>/hints")
@challenges_namespace.param("id", "A Challenge ID")
class ChallengeHints(Resource):
@admins_only
def get(self, challenge_id):
@@ -624,7 +686,6 @@ class ChallengeHints(Resource):
@challenges_namespace.route("/<challenge_id>/flags")
@challenges_namespace.param("id", "A Challenge ID")
class ChallengeFlags(Resource):
@admins_only
def get(self, challenge_id):

53
CTFd/api/v1/config.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_config, clear_standings
from CTFd.models import Configs, db
from CTFd.schemas.config import ConfigSchema
@@ -9,10 +13,39 @@ from CTFd.utils.decorators import admins_only
configs_namespace = Namespace("configs", description="Endpoint to retrieve Configs")
ConfigModel = sqlalchemy_to_pydantic(Configs)
class ConfigDetailedSuccessResponse(APIDetailedSuccessResponse):
data: ConfigModel
class ConfigListSuccessResponse(APIListSuccessResponse):
data: List[ConfigModel]
configs_namespace.schema_model(
"ConfigDetailedSuccessResponse", ConfigDetailedSuccessResponse.apidoc()
)
configs_namespace.schema_model(
"ConfigListSuccessResponse", ConfigListSuccessResponse.apidoc()
)
@configs_namespace.route("")
class ConfigList(Resource):
@admins_only
@configs_namespace.doc(
description="Endpoint to get Config objects in bulk",
responses={
200: ("Success", "ConfigListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
configs = Configs.query.all()
schema = ConfigSchema(many=True)
@@ -23,6 +56,16 @@ class ConfigList(Resource):
return {"success": True, "data": response.data}
@admins_only
@configs_namespace.doc(
description="Endpoint to get create a Config object",
responses={
200: ("Success", "ConfigDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
schema = ConfigSchema()
@@ -43,6 +86,10 @@ class ConfigList(Resource):
return {"success": True, "data": response.data}
@admins_only
@configs_namespace.doc(
description="Endpoint to get patch Config objects in bulk",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def patch(self):
req = request.get_json()
@@ -58,11 +105,13 @@ class ConfigList(Resource):
@configs_namespace.route("/<config_key>")
class Config(Resource):
@admins_only
# TODO: This returns weirdly structured data. It should more closely match ConfigDetailedSuccessResponse #1506
def get(self, config_key):
return {"success": True, "data": get_config(config_key)}
@admins_only
# TODO: This returns weirdly structured data. It should more closely match ConfigDetailedSuccessResponse #1506
def patch(self, config_key):
config = Configs.query.filter_by(key=config_key).first()
data = request.get_json()
@@ -89,6 +138,10 @@ class Config(Resource):
return {"success": True, "data": response.data}
@admins_only
@configs_namespace.doc(
description="Endpoint to delete a Config object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, config_key):
config = Configs.query.filter_by(key=config_key).first_or_404()

57
CTFd/api/v1/files.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.models import Files, db
from CTFd.schemas.files import FileSchema
from CTFd.utils import uploads
@@ -8,10 +12,39 @@ from CTFd.utils.decorators import admins_only
files_namespace = Namespace("files", description="Endpoint to retrieve Files")
FileModel = sqlalchemy_to_pydantic(Files)
class FileDetailedSuccessResponse(APIDetailedSuccessResponse):
data: FileModel
class FileListSuccessResponse(APIListSuccessResponse):
data: List[FileModel]
files_namespace.schema_model(
"FileDetailedSuccessResponse", FileDetailedSuccessResponse.apidoc()
)
files_namespace.schema_model(
"FileListSuccessResponse", FileListSuccessResponse.apidoc()
)
@files_namespace.route("")
class FilesList(Resource):
@admins_only
@files_namespace.doc(
description="Endpoint to get file objects in bulk",
responses={
200: ("Success", "FileListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
file_type = request.args.get("type")
files = Files.query.filter_by(type=file_type).all()
@@ -24,6 +57,16 @@ class FilesList(Resource):
return {"success": True, "data": response.data}
@admins_only
@files_namespace.doc(
description="Endpoint to get file objects in bulk",
responses={
200: ("Success", "FileDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
files = request.files.getlist("file")
# challenge_id
@@ -47,6 +90,16 @@ class FilesList(Resource):
@files_namespace.route("/<file_id>")
class FilesDetail(Resource):
@admins_only
@files_namespace.doc(
description="Endpoint to get a specific file object",
responses={
200: ("Success", "FileDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, file_id):
f = Files.query.filter_by(id=file_id).first_or_404()
schema = FileSchema()
@@ -58,6 +111,10 @@ class FilesDetail(Resource):
return {"success": True, "data": response.data}
@admins_only
@files_namespace.doc(
description="Endpoint to delete a file object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, file_id):
f = Files.query.filter_by(id=file_id).first_or_404()

67
CTFd/api/v1/flags.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.models import Flags, db
from CTFd.plugins.flags import FLAG_CLASSES, get_flag_class
from CTFd.schemas.flags import FlagSchema
@@ -8,10 +12,39 @@ from CTFd.utils.decorators import admins_only
flags_namespace = Namespace("flags", description="Endpoint to retrieve Flags")
FlagModel = sqlalchemy_to_pydantic(Flags)
class FlagDetailedSuccessResponse(APIDetailedSuccessResponse):
data: FlagModel
class FlagListSuccessResponse(APIListSuccessResponse):
data: List[FlagModel]
flags_namespace.schema_model(
"FlagDetailedSuccessResponse", FlagDetailedSuccessResponse.apidoc()
)
flags_namespace.schema_model(
"FlagListSuccessResponse", FlagListSuccessResponse.apidoc()
)
@flags_namespace.route("")
class FlagList(Resource):
@admins_only
@flags_namespace.doc(
description="Endpoint to list Flag objects in bulk",
responses={
200: ("Success", "FlagListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
flags = Flags.query.all()
schema = FlagSchema(many=True)
@@ -22,6 +55,16 @@ class FlagList(Resource):
return {"success": True, "data": response.data}
@admins_only
@flags_namespace.doc(
description="Endpoint to create a Flag object",
responses={
200: ("Success", "FlagDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
schema = FlagSchema()
@@ -62,6 +105,16 @@ class FlagTypes(Resource):
@flags_namespace.route("/<flag_id>")
class Flag(Resource):
@admins_only
@flags_namespace.doc(
description="Endpoint to get a specific Flag object",
responses={
200: ("Success", "FlagDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, flag_id):
flag = Flags.query.filter_by(id=flag_id).first_or_404()
schema = FlagSchema()
@@ -75,6 +128,10 @@ class Flag(Resource):
return {"success": True, "data": response.data}
@admins_only
@flags_namespace.doc(
description="Endpoint to delete a specific Flag object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, flag_id):
flag = Flags.query.filter_by(id=flag_id).first_or_404()
@@ -85,6 +142,16 @@ class Flag(Resource):
return {"success": True}
@admins_only
@flags_namespace.doc(
description="Endpoint to edit a specific Flag object",
responses={
200: ("Success", "FlagDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self, flag_id):
flag = Flags.query.filter_by(id=flag_id).first_or_404()
schema = FlagSchema()

0
CTFd/api/v1/helpers/__init__.py Normal file
View File

49
CTFd/api/v1/helpers/request.py Normal file
View File

@@ -0,0 +1,49 @@
from functools import wraps
from flask import request
from pydantic import create_model
ARG_LOCATIONS = {
"query": lambda: request.args,
"json": lambda: request.get_json(),
"form": lambda: request.form,
"headers": lambda: request.headers,
"cookies": lambda: request.cookies,
}
def validate_args(spec, location):
"""
A rough implementation of webargs using pydantic schemas. You can pass a
pydantic schema as spec or create it on the fly as follows:
@validate_args({"name": (str, None), "id": (int, None)}, location="query")
"""
if isinstance(spec, dict):
spec = create_model("", **spec)
schema = spec.schema()
props = schema.get("properties", {})
required = schema.get("required", [])
for k in props:
if k in required:
props[k]["required"] = True
props[k]["in"] = location
def decorator(func):
# Inject parameters information into the Flask-Restx apidoc attribute.
# Not really a good solution. See https://github.com/CTFd/CTFd/issues/1504
apidoc = getattr(func, "__apidoc__", {"params": {}})
apidoc["params"].update(props)
func.__apidoc__ = apidoc
@wraps(func)
def wrapper(*args, **kwargs):
data = ARG_LOCATIONS[location]()
loaded = spec(**data).dict(exclude_unset=True)
return func(*args, loaded, **kwargs)
return wrapper
return decorator

31
CTFd/api/v1/helpers/schemas.py Normal file
View File

@@ -0,0 +1,31 @@
from typing import Container, Type
from pydantic import BaseModel, create_model
from sqlalchemy.inspection import inspect
from sqlalchemy.orm.properties import ColumnProperty
def sqlalchemy_to_pydantic(
db_model: Type, *, exclude: Container[str] = []
) -> Type[BaseModel]:
"""
Mostly copied from https://github.com/tiangolo/pydantic-sqlalchemy
"""
mapper = inspect(db_model)
fields = {}
for attr in mapper.attrs:
if isinstance(attr, ColumnProperty):
if attr.columns:
column = attr.columns[0]
python_type = column.type.python_type
name = attr.key
if name in exclude:
continue
default = None
if column.default is None and not column.nullable:
default = ...
fields[name] = (python_type, default)
pydantic_model = create_model(
db_model.__name__, **fields # type: ignore
)
return pydantic_model

67
CTFd/api/v1/hints.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.models import Hints, HintUnlocks, db
from CTFd.schemas.hints import HintSchema
from CTFd.utils.decorators import admins_only, authed_only, during_ctf_time_only
@@ -8,10 +12,39 @@ from CTFd.utils.user import get_current_user, is_admin
hints_namespace = Namespace("hints", description="Endpoint to retrieve Hints")
HintModel = sqlalchemy_to_pydantic(Hints)
class HintDetailedSuccessResponse(APIDetailedSuccessResponse):
data: HintModel
class HintListSuccessResponse(APIListSuccessResponse):
data: List[HintModel]
hints_namespace.schema_model(
"HintDetailedSuccessResponse", HintDetailedSuccessResponse.apidoc()
)
hints_namespace.schema_model(
"HintListSuccessResponse", HintListSuccessResponse.apidoc()
)
@hints_namespace.route("")
class HintList(Resource):
@admins_only
@hints_namespace.doc(
description="Endpoint to list Hint objects in bulk",
responses={
200: ("Success", "HintListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
hints = Hints.query.all()
response = HintSchema(many=True).dump(hints)
@@ -22,6 +55,16 @@ class HintList(Resource):
return {"success": True, "data": response.data}
@admins_only
@hints_namespace.doc(
description="Endpoint to create a Hint object",
responses={
200: ("Success", "HintDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
schema = HintSchema("admin")
@@ -42,6 +85,16 @@ class HintList(Resource):
class Hint(Resource):
@during_ctf_time_only
@authed_only
@hints_namespace.doc(
description="Endpoint to get a specific Hint object",
responses={
200: ("Success", "HintDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, hint_id):
user = get_current_user()
hint = Hints.query.filter_by(id=hint_id).first_or_404()
@@ -67,6 +120,16 @@ class Hint(Resource):
return {"success": True, "data": response.data}
@admins_only
@hints_namespace.doc(
description="Endpoint to edit a specific Hint object",
responses={
200: ("Success", "HintDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self, hint_id):
hint = Hints.query.filter_by(id=hint_id).first_or_404()
req = request.get_json()
@@ -85,6 +148,10 @@ class Hint(Resource):
return {"success": True, "data": response.data}
@admins_only
@hints_namespace.doc(
description="Endpoint to delete a specific Tag object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, hint_id):
hint = Hints.query.filter_by(id=hint_id).first_or_404()
db.session.delete(hint)

58
CTFd/api/v1/notifications.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import current_app, request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.models import Notifications, db
from CTFd.schemas.notifications import NotificationSchema
from CTFd.utils.decorators import admins_only
@@ -9,9 +13,39 @@ notifications_namespace = Namespace(
"notifications", description="Endpoint to retrieve Notifications"
)
NotificationModel = sqlalchemy_to_pydantic(Notifications)
TransientNotificationModel = sqlalchemy_to_pydantic(Notifications, exclude=["id"])
class NotificationDetailedSuccessResponse(APIDetailedSuccessResponse):
data: NotificationModel
class NotificationListSuccessResponse(APIListSuccessResponse):
data: List[NotificationModel]
notifications_namespace.schema_model(
"NotificationDetailedSuccessResponse", NotificationDetailedSuccessResponse.apidoc()
)
notifications_namespace.schema_model(
"NotificationListSuccessResponse", NotificationListSuccessResponse.apidoc()
)
@notifications_namespace.route("")
class NotificantionList(Resource):
@notifications_namespace.doc(
description="Endpoint to get notification objects in bulk",
responses={
200: ("Success", "NotificationListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
notifications = Notifications.query.all()
schema = NotificationSchema(many=True)
@@ -21,6 +55,16 @@ class NotificantionList(Resource):
return {"success": True, "data": result.data}
@admins_only
@notifications_namespace.doc(
description="Endpoint to create a notification object",
responses={
200: ("Success", "NotificationDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
@@ -49,6 +93,16 @@ class NotificantionList(Resource):
@notifications_namespace.route("/<notification_id>")
@notifications_namespace.param("notification_id", "A Notification ID")
class Notification(Resource):
@notifications_namespace.doc(
description="Endpoint to get a specific notification object",
responses={
200: ("Success", "NotificationDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, notification_id):
notif = Notifications.query.filter_by(id=notification_id).first_or_404()
schema = NotificationSchema()
@@ -59,6 +113,10 @@ class Notification(Resource):
return {"success": True, "data": response.data}
@admins_only
@notifications_namespace.doc(
description="Endpoint to delete a notification object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, notification_id):
notif = Notifications.query.filter_by(id=notification_id).first_or_404()
db.session.delete(notif)

85
CTFd/api/v1/pages.py
View File

@@ -1,6 +1,11 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.request import validate_args
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_pages
from CTFd.models import Pages, db
from CTFd.schemas.pages import PageSchema
@@ -9,11 +14,56 @@ from CTFd.utils.decorators import admins_only
pages_namespace = Namespace("pages", description="Endpoint to retrieve Pages")
PageModel = sqlalchemy_to_pydantic(Pages)
TransientPageModel = sqlalchemy_to_pydantic(Pages, exclude=["id"])
class PageDetailedSuccessResponse(APIDetailedSuccessResponse):
data: PageModel
class PageListSuccessResponse(APIListSuccessResponse):
data: List[PageModel]
pages_namespace.schema_model(
"PageDetailedSuccessResponse", PageDetailedSuccessResponse.apidoc()
)
pages_namespace.schema_model(
"PageListSuccessResponse", PageListSuccessResponse.apidoc()
)
@pages_namespace.route("")
@pages_namespace.doc(
responses={200: "Success", 400: "An error occured processing your data"}
)
class PageList(Resource):
@admins_only
def get(self):
pages = Pages.query.all()
@pages_namespace.doc(
description="Endpoint to get page objects in bulk",
responses={
200: ("Success", "PageListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
@validate_args(
{
"id": (int, None),
"title": (str, None),
"route": (str, None),
"draft": (bool, None),
"hidden": (bool, None),
"auth_required": (bool, None),
},
location="query",
)
def get(self, query):
pages = Pages.query.filter_by(**query).all()
schema = PageSchema(exclude=["content"], many=True)
response = schema.dump(pages)
if response.errors:
@@ -22,8 +72,19 @@ class PageList(Resource):
return {"success": True, "data": response.data}
@admins_only
def post(self):
req = request.get_json()
@pages_namespace.doc(
description="Endpoint to create a page object",
responses={
200: ("Success", "PageDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
@validate_args(TransientPageModel, location="json")
def post(self, json_args):
req = json_args
schema = PageSchema()
response = schema.load(req)
@@ -42,8 +103,19 @@ class PageList(Resource):
@pages_namespace.route("/<page_id>")
@pages_namespace.doc(
params={"page_id": "ID of a page object"},
responses={
200: ("Success", "PageDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
class PageDetail(Resource):
@admins_only
@pages_namespace.doc(description="Endpoint to read a page object")
def get(self, page_id):
page = Pages.query.filter_by(id=page_id).first_or_404()
schema = PageSchema()
@@ -55,6 +127,7 @@ class PageDetail(Resource):
return {"success": True, "data": response.data}
@admins_only
@pages_namespace.doc(description="Endpoint to edit a page object")
def patch(self, page_id):
page = Pages.query.filter_by(id=page_id).first_or_404()
req = request.get_json()
@@ -75,6 +148,10 @@ class PageDetail(Resource):
return {"success": True, "data": response.data}
@admins_only
@pages_namespace.doc(
description="Endpoint to delete a page object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, page_id):
page = Pages.query.filter_by(id=page_id).first_or_404()
db.session.delete(page)

60
CTFd/api/v1/schemas/__init__.py Normal file
View File

@@ -0,0 +1,60 @@
from typing import Any, List, Optional
from pydantic import BaseModel
class APISimpleSuccessResponse(BaseModel):
success: bool = True
class APIDetailedSuccessResponse(APISimpleSuccessResponse):
data: Optional[Any]
@classmethod
def apidoc(cls):
"""
Helper to inline references from the generated schema
"""
schema = cls.schema()
try:
key = schema["properties"]["data"]["$ref"]
ref = key.split("/").pop()
definition = schema["definitions"][ref]
schema["properties"]["data"] = definition
del schema["definitions"][ref]
if bool(schema["definitions"]) is False:
del schema["definitions"]
except KeyError:
pass
return schema
class APIListSuccessResponse(APIDetailedSuccessResponse):
data: Optional[List[Any]]
@classmethod
def apidoc(cls):
"""
Helper to inline references from the generated schema
"""
schema = cls.schema()
try:
key = schema["properties"]["data"]["items"]["$ref"]
ref = key.split("/").pop()
definition = schema["definitions"][ref]
schema["properties"]["data"]["items"] = definition
del schema["definitions"][ref]
if bool(schema["definitions"]) is False:
del schema["definitions"]
except KeyError:
pass
return schema
class APISimpleErrorResponse(BaseModel):
success: bool = False
errors: Optional[List[str]]

70
CTFd/api/v1/submissions.py
View File

@@ -1,6 +1,11 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.request import validate_args
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_standings
from CTFd.models import Submissions, db
from CTFd.schemas.submissions import SubmissionSchema
@@ -10,10 +15,40 @@ submissions_namespace = Namespace(
"submissions", description="Endpoint to retrieve Submission"
)
SubmissionModel = sqlalchemy_to_pydantic(Submissions)
TransientSubmissionModel = sqlalchemy_to_pydantic(Submissions, exclude=["id"])
class SubmissionDetailedSuccessResponse(APIDetailedSuccessResponse):
data: SubmissionModel
class SubmissionListSuccessResponse(APIListSuccessResponse):
data: List[SubmissionModel]
submissions_namespace.schema_model(
"SubmissionDetailedSuccessResponse", SubmissionDetailedSuccessResponse.apidoc()
)
submissions_namespace.schema_model(
"SubmissionListSuccessResponse", SubmissionListSuccessResponse.apidoc()
)
@submissions_namespace.route("")
class SubmissionsList(Resource):
@admins_only
@submissions_namespace.doc(
description="Endpoint to get submission objects in bulk",
responses={
200: ("Success", "SubmissionListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
args = request.args.to_dict()
schema = SubmissionSchema(many=True)
@@ -30,8 +65,19 @@ class SubmissionsList(Resource):
return {"success": True, "data": response.data}
@admins_only
def post(self):
req = request.get_json()
@submissions_namespace.doc(
description="Endpoint to create a submission object. Users should interact with the attempt endpoint to submit flags.",
responses={
200: ("Success", "SubmissionListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
@validate_args(TransientSubmissionModel, location="json")
def post(self, json_args):
req = json_args
Model = Submissions.get_child(type=req.get("type"))
schema = SubmissionSchema(instance=Model())
response = schema.load(req)
@@ -54,6 +100,16 @@ class SubmissionsList(Resource):
@submissions_namespace.param("submission_id", "A Submission ID")
class Submission(Resource):
@admins_only
@submissions_namespace.doc(
description="Endpoint to get submission objects in bulk",
responses={
200: ("Success", "SubmissionDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, submission_id):
submission = Submissions.query.filter_by(id=submission_id).first_or_404()
schema = SubmissionSchema()
@@ -65,6 +121,16 @@ class Submission(Resource):
return {"success": True, "data": response.data}
@admins_only
@submissions_namespace.doc(
description="Endpoint to get submission objects in bulk",
responses={
200: ("Success", "APISimpleSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def delete(self, submission_id):
submission = Submissions.query.filter_by(id=submission_id).first_or_404()
db.session.delete(submission)

65
CTFd/api/v1/tags.py
View File

@@ -1,16 +1,47 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.models import Tags, db
from CTFd.schemas.tags import TagSchema
from CTFd.utils.decorators import admins_only
tags_namespace = Namespace("tags", description="Endpoint to retrieve Tags")
TagModel = sqlalchemy_to_pydantic(Tags)
class TagDetailedSuccessResponse(APIDetailedSuccessResponse):
data: TagModel
class TagListSuccessResponse(APIListSuccessResponse):
data: List[TagModel]
tags_namespace.schema_model(
"TagDetailedSuccessResponse", TagDetailedSuccessResponse.apidoc()
)
tags_namespace.schema_model("TagListSuccessResponse", TagListSuccessResponse.apidoc())
@tags_namespace.route("")
class TagList(Resource):
@admins_only
@tags_namespace.doc(
description="Endpoint to list Tag objects in bulk",
responses={
200: ("Success", "TagListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
# TODO: Filter by challenge_id
tags = Tags.query.all()
@@ -23,6 +54,16 @@ class TagList(Resource):
return {"success": True, "data": response.data}
@admins_only
@tags_namespace.doc(
description="Endpoint to create a Tag object",
responses={
200: ("Success", "TagDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
schema = TagSchema()
@@ -44,6 +85,16 @@ class TagList(Resource):
@tags_namespace.param("tag_id", "A Tag ID")
class Tag(Resource):
@admins_only
@tags_namespace.doc(
description="Endpoint to get a specific Tag object",
responses={
200: ("Success", "TagDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, tag_id):
tag = Tags.query.filter_by(id=tag_id).first_or_404()
@@ -55,6 +106,16 @@ class Tag(Resource):
return {"success": True, "data": response.data}
@admins_only
@tags_namespace.doc(
description="Endpoint to edit a specific Tag object",
responses={
200: ("Success", "TagDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self, tag_id):
tag = Tags.query.filter_by(id=tag_id).first_or_404()
schema = TagSchema()
@@ -72,6 +133,10 @@ class Tag(Resource):
return {"success": True, "data": response.data}
@admins_only
@tags_namespace.doc(
description="Endpoint to delete a specific Tag object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, tag_id):
tag = Tags.query.filter_by(id=tag_id).first_or_404()
db.session.delete(tag)

87
CTFd/api/v1/teams.py
View File

@@ -1,8 +1,11 @@
import copy
from typing import List
from flask import abort, request, session
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_standings, clear_team_session, clear_user_session
from CTFd.models import Awards, Submissions, Teams, Unlocks, Users, db
from CTFd.schemas.awards import AwardSchema
@@ -17,10 +20,40 @@ from CTFd.utils.user import get_current_team, get_current_user_type, is_admin
teams_namespace = Namespace("teams", description="Endpoint to retrieve Teams")
TeamModel = sqlalchemy_to_pydantic(Teams)
TransientTeamModel = sqlalchemy_to_pydantic(Teams, exclude=["id"])
class TeamDetailedSuccessResponse(APIDetailedSuccessResponse):
data: TeamModel
class TeamListSuccessResponse(APIListSuccessResponse):
data: List[TeamModel]
teams_namespace.schema_model(
"TeamDetailedSuccessResponse", TeamDetailedSuccessResponse.apidoc()
)
teams_namespace.schema_model(
"TeamListSuccessResponse", TeamListSuccessResponse.apidoc()
)
@teams_namespace.route("")
class TeamList(Resource):
@check_account_visibility
@teams_namespace.doc(
description="Endpoint to get Team objects in bulk",
responses={
200: ("Success", "TeamListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
if is_admin() and request.args.get("view") == "admin":
teams = Teams.query.filter_by()
@@ -38,6 +71,16 @@ class TeamList(Resource):
return {"success": True, "data": response.data}
@admins_only
@teams_namespace.doc(
description="Endpoint to create a Team object",
responses={
200: ("Success", "TeamDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
user_type = get_current_user_type()
@@ -63,6 +106,16 @@ class TeamList(Resource):
@teams_namespace.param("team_id", "Team ID")
class TeamPublic(Resource):
@check_account_visibility
@teams_namespace.doc(
description="Endpoint to get a specific Team object",
responses={
200: ("Success", "TeamDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, team_id):
team = Teams.query.filter_by(id=team_id).first_or_404()
@@ -82,6 +135,16 @@ class TeamPublic(Resource):
return {"success": True, "data": response.data}
@admins_only
@teams_namespace.doc(
description="Endpoint to edit a specific Team object",
responses={
200: ("Success", "TeamDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self, team_id):
team = Teams.query.filter_by(id=team_id).first_or_404()
data = request.get_json()
@@ -104,6 +167,10 @@ class TeamPublic(Resource):
return {"success": True, "data": response.data}
@admins_only
@teams_namespace.doc(
description="Endpoint to delete a specific Team object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, team_id):
team = Teams.query.filter_by(id=team_id).first_or_404()
team_id = team.id
@@ -128,6 +195,16 @@ class TeamPublic(Resource):
class TeamPrivate(Resource):
@authed_only
@require_team
@teams_namespace.doc(
description="Endpoint to get the current user's Team object",
responses={
200: ("Success", "TeamDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
team = get_current_team()
response = TeamSchema(view="self").dump(team)
@@ -141,6 +218,16 @@ class TeamPrivate(Resource):
@authed_only
@require_team
@teams_namespace.doc(
description="Endpoint to edit the current user's Team object",
responses={
200: ("Success", "TeamDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self):
team = get_current_team()
if team.captain_id != session["id"]:

66
CTFd/api/v1/tokens.py
View File

@@ -1,8 +1,11 @@
import datetime
from typing import List
from flask import request, session
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.models import Tokens, db
from CTFd.schemas.tokens import TokenSchema
from CTFd.utils.decorators import authed_only, require_verified_emails
@@ -11,11 +14,50 @@ from CTFd.utils.user import get_current_user, get_current_user_type, is_admin
tokens_namespace = Namespace("tokens", description="Endpoint to retrieve Tokens")
TokenModel = sqlalchemy_to_pydantic(Tokens)
ValuelessTokenModel = sqlalchemy_to_pydantic(Tokens, exclude=["value"])
class TokenDetailedSuccessResponse(APIDetailedSuccessResponse):
data: TokenModel
class ValuelessTokenDetailedSuccessResponse(APIDetailedSuccessResponse):
data: ValuelessTokenModel
class TokenListSuccessResponse(APIListSuccessResponse):
data: List[TokenModel]
tokens_namespace.schema_model(
"TokenDetailedSuccessResponse", TokenDetailedSuccessResponse.apidoc()
)
tokens_namespace.schema_model(
"ValuelessTokenDetailedSuccessResponse",
ValuelessTokenDetailedSuccessResponse.apidoc(),
)
tokens_namespace.schema_model(
"TokenListSuccessResponse", TokenListSuccessResponse.apidoc()
)
@tokens_namespace.route("")
class TokenList(Resource):
@require_verified_emails
@authed_only
@tokens_namespace.doc(
description="Endpoint to get token objects in bulk",
responses={
200: ("Success", "TokenListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
user = get_current_user()
tokens = Tokens.query.filter_by(user_id=user.id)
@@ -30,6 +72,16 @@ class TokenList(Resource):
@require_verified_emails
@authed_only
@tokens_namespace.doc(
description="Endpoint to create a token object",
responses={
200: ("Success", "TokenDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
expiration = req.get("expiration")
@@ -54,6 +106,16 @@ class TokenList(Resource):
class TokenDetail(Resource):
@require_verified_emails
@authed_only
@tokens_namespace.doc(
description="Endpoint to get an existing token object",
responses={
200: ("Success", "ValuelessTokenDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, token_id):
if is_admin():
token = Tokens.query.filter_by(id=token_id).first_or_404()
@@ -73,6 +135,10 @@ class TokenDetail(Resource):
@require_verified_emails
@authed_only
@tokens_namespace.doc(
description="Endpoint to delete an existing token object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, token_id):
if is_admin():
token = Tokens.query.filter_by(id=token_id).first_or_404()

44
CTFd/api/v1/unlocks.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_standings
from CTFd.models import Unlocks, db, get_class_by_tablename
from CTFd.schemas.awards import AwardSchema
@@ -15,10 +19,40 @@ from CTFd.utils.user import get_current_user
unlocks_namespace = Namespace("unlocks", description="Endpoint to retrieve Unlocks")
UnlockModel = sqlalchemy_to_pydantic(Unlocks)
TransientUnlockModel = sqlalchemy_to_pydantic(Unlocks, exclude=["id"])
class UnlockDetailedSuccessResponse(APIDetailedSuccessResponse):
data: UnlockModel
class UnlockListSuccessResponse(APIListSuccessResponse):
data: List[UnlockModel]
unlocks_namespace.schema_model(
"UnlockDetailedSuccessResponse", UnlockDetailedSuccessResponse.apidoc()
)
unlocks_namespace.schema_model(
"UnlockListSuccessResponse", UnlockListSuccessResponse.apidoc()
)
@unlocks_namespace.route("")
class UnlockList(Resource):
@admins_only
@unlocks_namespace.doc(
description="Endpoint to get unlock objects in bulk",
responses={
200: ("Success", "UnlockListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
hints = Unlocks.query.all()
schema = UnlockSchema()
@@ -32,6 +66,16 @@ class UnlockList(Resource):
@during_ctf_time_only
@require_verified_emails
@authed_only
@unlocks_namespace.doc(
description="Endpoint to create an unlock object. Used to unlock hints.",
responses={
200: ("Success", "UnlockDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def post(self):
req = request.get_json()
user = get_current_user()

98
CTFd/api/v1/users.py
View File

@@ -1,6 +1,10 @@
from typing import List
from flask import abort, request
from flask_restx import Namespace, Resource
from CTFd.api.v1.helpers.schemas import sqlalchemy_to_pydantic
from CTFd.api.v1.schemas import APIDetailedSuccessResponse, APIListSuccessResponse
from CTFd.cache import clear_standings, clear_user_session
from CTFd.models import (
Awards,
@@ -28,9 +32,40 @@ from CTFd.utils.user import get_current_user, get_current_user_type, is_admin
users_namespace = Namespace("users", description="Endpoint to retrieve Users")
UserModel = sqlalchemy_to_pydantic(Users)
TransientUserModel = sqlalchemy_to_pydantic(Users, exclude=["id"])
class UserDetailedSuccessResponse(APIDetailedSuccessResponse):
data: UserModel
class UserListSuccessResponse(APIListSuccessResponse):
data: List[UserModel]
users_namespace.schema_model(
"UserDetailedSuccessResponse", UserDetailedSuccessResponse.apidoc()
)
users_namespace.schema_model(
"UserListSuccessResponse", UserListSuccessResponse.apidoc()
)
@users_namespace.route("")
class UserList(Resource):
@check_account_visibility
@users_namespace.doc(
description="Endpoint to get User objects in bulk",
responses={
200: ("Success", "UserListSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
if is_admin() and request.args.get("view") == "admin":
users = Users.query.filter_by()
@@ -44,12 +79,21 @@ class UserList(Resource):
return {"success": True, "data": response.data}
@users_namespace.doc()
@admins_only
@users_namespace.doc(
description="Endpoint to create a User object",
responses={
200: ("Success", "UserDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
params={
"notify": "Whether to send the created user an email with their credentials"
}
},
)
@admins_only
def post(self):
req = request.get_json()
schema = UserSchema("admin")
@@ -79,6 +123,16 @@ class UserList(Resource):
@users_namespace.param("user_id", "User ID")
class UserPublic(Resource):
@check_account_visibility
@users_namespace.doc(
description="Endpoint to get a specific User object",
responses={
200: ("Success", "UserDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self, user_id):
user = Users.query.filter_by(id=user_id).first_or_404()
@@ -97,6 +151,16 @@ class UserPublic(Resource):
return {"success": True, "data": response.data}
@admins_only
@users_namespace.doc(
description="Endpoint to edit a specific User object",
responses={
200: ("Success", "UserDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self, user_id):
user = Users.query.filter_by(id=user_id).first_or_404()
data = request.get_json()
@@ -118,6 +182,10 @@ class UserPublic(Resource):
return {"success": True, "data": response}
@admins_only
@users_namespace.doc(
description="Endpoint to delete a specific User object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
def delete(self, user_id):
Notifications.query.filter_by(user_id=user_id).delete()
Awards.query.filter_by(user_id=user_id).delete()
@@ -138,6 +206,16 @@ class UserPublic(Resource):
@users_namespace.route("/me")
class UserPrivate(Resource):
@authed_only
@users_namespace.doc(
description="Endpoint to get the User object for the current user",
responses={
200: ("Success", "UserDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def get(self):
user = get_current_user()
response = UserSchema("self").dump(user).data
@@ -146,6 +224,16 @@ class UserPrivate(Resource):
return {"success": True, "data": response}
@authed_only
@users_namespace.doc(
description="Endpoint to edit the User object for the current user",
responses={
200: ("Success", "UserDetailedSuccessResponse"),
400: (
"An error occured processing the provided or stored data",
"APISimpleErrorResponse",
),
},
)
def patch(self):
user = get_current_user()
data = request.get_json()
@@ -294,6 +382,10 @@ class UserPublicAwards(Resource):
@users_namespace.param("user_id", "User ID")
class UserEmails(Resource):
@admins_only
@users_namespace.doc(
description="Endpoint to email a User object",
responses={200: ("Success", "APISimpleSuccessResponse")},
)
@ratelimit(method="POST", limit=10, interval=60)
def post(self, user_id):
req = request.get_json()
@@ -314,4 +406,4 @@ class UserEmails(Resource):
result, response = sendmail(addr=user.email, text=text)
return {"success": result, "data": {}}
return {"success": result}

1
requirements.txt
View File

@@ -22,6 +22,7 @@ flask-marshmallow==0.10.1
marshmallow-sqlalchemy==0.17.0
boto3==1.13.9
marshmallow==2.20.2
pydantic==1.5.1
lxml==4.5.1
html5lib==1.0.1
WTForms==2.3.1