[rewrite] Removed old code, updated .vscode

This commit is contained in:
Thastertyn 2025-02-17 21:59:41 +01:00
parent 3c005dbfa4
commit 3b0336d9b4
32 changed files with 4 additions and 1105 deletions

View File

@ -1,9 +0,0 @@
from flask import Blueprint
bp_errors = Blueprint('errors', __name__)
bp = Blueprint('api', __name__)
bp_product = Blueprint('products', __name__, url_prefix="/products")
bp_user = Blueprint('user', __name__, url_prefix="/user")
bp_cart = Blueprint('cart', __name__, url_prefix="/cart")
from . import routes

View File

@ -1,15 +0,0 @@
from app.api.routes.user import (
register_route,
login_route,
logout_route,
update_route,
delete_route,
)
from app.api.routes.product import (
product_create_route,
product_delete_route,
product_info_route,
product_page_route,
)
from app.api.routes import main_routes, error_routes, cart_routes

View File

@ -1,79 +0,0 @@
from flask import jsonify, abort, request
from flask_jwt_extended import jwt_required, get_jwt_identity
from app.doc.cart_swag import (
show_cart_swagger,
add_to_cart_swagger,
remove_from_cart_swagger,
update_count_in_cart_swagger,
purchase_swagger,
)
from flasgger import swag_from
from app.api import bp_cart
from app.services.cart_service import CartService
@bp_cart.route("", methods=["GET"])
@jwt_required()
@swag_from(show_cart_swagger)
def show_cart():
user_id = get_jwt_identity()
result, status_code = CartService.show_cart(user_id)
return result, status_code
@bp_cart.route("/add/<int:product_id>", methods=["PUT"])
@jwt_required()
@swag_from(add_to_cart_swagger)
def add_to_cart(product_id: int):
user_id = get_jwt_identity()
count = request.args.get("count", default=1, type=int)
if count < 1:
return abort(400)
result, status_code = CartService.add_to_cart(user_id, product_id, count)
return result, status_code
@bp_cart.route("/remove/<int:product_id>", methods=["DELETE"])
@jwt_required()
@swag_from(remove_from_cart_swagger)
def remove_from_cart(product_id: int):
user_id = get_jwt_identity()
result, status_code = CartService.delete_from_cart(user_id, product_id)
return result, status_code
@bp_cart.route("/update/<int:product_id>", methods=["PUT"])
@jwt_required()
@swag_from(update_count_in_cart_swagger)
def update_count_in_cart(product_id: int):
user_id = get_jwt_identity()
count = request.args.get("count", type=int)
if not count:
return abort(400)
result, status_code = CartService.update_count(user_id, product_id, count)
return result, status_code
@bp_cart.route("/purchase", methods=["GET"])
@jwt_required()
@swag_from(purchase_swagger)
def purchase():
user_id = get_jwt_identity()
result, status_code = CartService.purchase(user_id)
return result, status_code

View File

@ -1,38 +0,0 @@
from app.api import bp_errors
@bp_errors.app_errorhandler(400)
def bad_request(e):
return {
"msg": "The request was incorrectly formatted, or contained invalid data"
}, 400
@bp_errors.app_errorhandler(401)
def unauthorized(e):
return {"msg": "Failed to authorize the request"}, 401
@bp_errors.app_errorhandler(403)
def forbidden(e):
return {"msg": "You shall not pass"}, 403
@bp_errors.app_errorhandler(404)
def not_found(e):
return {"msg": "The requested resource was not found"}, 404
@bp_errors.app_errorhandler(405)
def method_not_allowed(e):
return {"msg": "The method used is not allowed in current context"}, 405
@bp_errors.app_errorhandler(500)
def internal_error(e):
return {"msg": "An error occurred on he server"}, 500
@bp_errors.app_errorhandler(501)
def unimplemented_error(e):
return {"msg": "This function has not been implemented yet. Check back soon!"}, 501

View File

@ -1,12 +0,0 @@
from flask import jsonify
from flasgger import swag_from
from app.doc.root_swag import root_swagger
from app.api import bp
@bp.route("/")
@swag_from(root_swagger)
def hello():
return jsonify({"message": "Hello, Flask!"})

View File

@ -1,33 +0,0 @@
from flask import jsonify, abort, request
from flask_jwt_extended import jwt_required, get_jwt_identity
from app.doc.product_swag import create_product_swagger
from flasgger import swag_from
from app.api import bp_product
from app.services.product import product_create_service
@bp_product.route("/create", methods=["POST"])
@swag_from(create_product_swagger)
@jwt_required()
def create_product_listing():
user_id = get_jwt_identity()
name = request.json.get("name")
price = request.json.get("price")
if name is None or price is None:
return abort(400)
float_price = float(price)
if not isinstance(float_price, float):
return abort(400)
result, status_code = product_create_service.create_product(
user_id, name, float_price
)
return jsonify(result), status_code

View File

@ -1,19 +0,0 @@
from flask import jsonify, abort, request
from flask_jwt_extended import jwt_required, get_jwt_identity
from flasgger import swag_from
from app.api import bp_product
from app.services.product import product_delete_service
@bp_product.route("/<int:product_id>/delete", methods=["DELETE"])
@jwt_required()
def delete_product(product_id: int):
user_id = get_jwt_identity()
result, status_code = product_delete_service.delete_product(user_id, product_id)
return jsonify(result), status_code

View File

@ -1,25 +0,0 @@
from flask import jsonify, request
from app.doc.product_swag import get_product_info_swagger
from flasgger import swag_from
from app.api import bp_product
from app.services.product import product_info_service
@bp_product.route("/<int:product_id>", methods=["GET"])
@swag_from(get_product_info_swagger)
def get_product_info(product_id: int):
fields = ["name", "price", "image", "image_name", "seller"]
fields_param = request.args.get("fields")
fields_param_list = fields_param.split(",") if fields_param else fields
common_fields = list(set(fields) & set(fields_param_list))
result, status_code = product_info_service.product_info(product_id)
return jsonify(result), status_code

View File

@ -1,22 +0,0 @@
from flask import jsonify, abort, request
from app.doc.product_swag import get_products_swagger
from flasgger import swag_from
from app.api import bp_product
from app.services.product import product_list_service
@bp_product.route("", methods=["GET"])
@swag_from(get_products_swagger)
def get_products():
page = request.args.get("page", default=0, type=int)
if page < 0:
return abort(400)
result, status_code = product_list_service.product_list(page)
return jsonify(result), status_code

View File

@ -1,22 +0,0 @@
from app.api import bp_user
from flask_jwt_extended import jwt_required, get_jwt_identity, get_jwt
from flask import request, abort
from flasgger import swag_from
from app.doc.user_swag import delete_swagger
from app.services.user import delete_service, logout_service
@bp_user.route("/delete", methods=["DELETE"])
@swag_from(delete_swagger)
@jwt_required()
def delete_user():
user_id = get_jwt_identity()
result, status_code = delete_service.delete_user(user_id)
jwt = get_jwt()
logout_service.logout(jwt, user_id, True)
return result, status_code

View File

@ -1,33 +0,0 @@
from app.api import bp_user
from flask import request, jsonify
from flasgger import swag_from
import app.messages.api_responses.user_responses as response
import app.messages.api_errors as errors
from app.doc.user_swag import login_swagger
from app.services.user import login_service
@bp_user.route("/login", methods=["POST"])
@swag_from(login_swagger)
def login():
data = request.get_json()
if not data:
result, status_code = errors.NOT_JSON
return jsonify(result), status_code
required_fields = ["username", "password"]
missing_fields = [field for field in required_fields if field not in data]
if missing_fields:
result, status_code = errors.MISSING_FIELDS(missing_fields)
return jsonify(result), status_code
username = data["username"]
password = data["password"]
result, status_code = login_service.login(username, password)
return result, status_code

View File

@ -1,20 +0,0 @@
from app.api import bp_user
from flasgger import swag_from
from flask_jwt_extended import get_jwt_identity, jwt_required, get_jwt
from app.doc.user_swag import logout_swagger
from app.services.user import logout_service
@bp_user.route("/logout", methods=["DELETE"])
@swag_from(logout_swagger)
@jwt_required()
def logout():
jwt = get_jwt()
user_id = get_jwt_identity()
result, status_code = logout_service.logout(jwt, user_id, True)
return result, status_code

View File

@ -1,39 +0,0 @@
from app.api import bp_user
from flask import request, jsonify
from app.services.user import register_service
from app.doc.user_swag import register_swagger
import app.messages.api_responses.user_responses as response
import app.messages.api_errors as errors
from flasgger import swag_from
@bp_user.route("/register", methods=["POST"])
@swag_from(register_swagger)
def register():
data = request.get_json()
if not data:
result, status_code = errors.NOT_JSON
return jsonify(result), status_code
required_fields = ["username", "displayname", "email", "password"]
missing_fields = [field for field in required_fields if field not in data]
if missing_fields:
result, status_code = errors.MISSING_FIELDS(missing_fields)
return jsonify(result), status_code
username = data["username"]
displayname = data["displayname"]
email = data["email"]
password = data["password"]
result, status_code = register_service.register(
username, displayname, email, password
)
return jsonify(result), status_code

View File

@ -1,40 +0,0 @@
from app.api import bp_user
from flask import request, jsonify
from flask_jwt_extended import jwt_required, get_jwt_identity, get_jwt
from flasgger import swag_from
import app.messages.api_errors as errors
from app.doc.user_swag import update_swagger
from app.services.user import logout_service, update_user_service
@bp_user.route("/update", methods=["PUT"])
@swag_from(update_swagger)
@jwt_required()
def update_user():
data = request.get_json()
possible_fields = ["new_username", "new_displayname", "new_email", "new_password"]
selected_fields = [field for field in possible_fields if field in data]
if not selected_fields:
result, status_code = errors.NO_FIELD_PROVIDED(possible_fields)
return jsonify(result), status_code
user_id = get_jwt_identity()
new_username = data.get("new_username")
new_displayname = data.get("new_displayname")
new_email = data.get("new_email")
new_password = data.get("new_password")
result, status_code = update_user_service.update_user(user_id, new_username, new_displayname, new_email, new_password)
if status_code < 300:
jwt = get_jwt()
logout_service.logout(jwt, user_id, False)
return result, status_code

View File

@ -1,118 +0,0 @@
from typing import Optional
from app.extensions import db_connection
from app.models.product_model import Product
def fetch_products(page: int = 0) -> Optional[list[Product]]:
cursor = db_connection.cursor(dictionary=True)
offset = 10 * page
cursor.execute(
"select product.id, user.displayname as seller, product.name, product.price_pc from product inner join user on user.id = product.seller_id order by product.id limit 10 offset %s",
(offset,),
)
results = cursor.fetchall()
if len(results) < 1:
return None
result_products: list[Product] = []
for row in results:
result_products.append(
Product(
product_id=row["id"],
seller_id=row["seller_id"],
name=row["name"],
price=row["price"],
creation_date=row["creation_date"],
)
)
return result_products
def fetch_product_by_id(product_id: int) -> Optional[Product]:
"""
Fetches specific product info
:param product_id: ID of product to be updated.
:type product_id: int
"""
cursor = db_connection.cursor(dictionary=True)
cursor.execute("select * from product where id = %s", (product_id,))
result = cursor.fetchone()
if cursor.rowcount != 1:
return None
result_product = Product(
product_id=result["id"],
seller_id=result["seller_id"],
name=result["name"],
price=result["price"],
creation_date=result["creation_date"],
)
return result_product
def fetch_product_extended_by_id(product_id: int) -> Optional[Product]:
"""
Fetches specific product info including the seller n
:param product_id: ID of product to be updated.
:type product_id: int
"""
cursor = db_connection.cursor(dictionary=True)
cursor.execute("select * from product inner join user on user.id = product.seller_id where product.id = %s", (product_id,))
result = cursor.fetchone()
if cursor.rowcount != 1:
return None
result_product = Product(
product_id=result["id"],
seller_id=result["seller_id"],
seller_name=result["displayname"],
name=result["name"],
price=result["price"],
creation_date=result["creation_date"],
)
return result_product
def insert_product(product: Product):
"""
Creates a new product listing
:param seller_id: User ID
:type seller_id: str
:param name: New product's name
:type name: str
:param price: New product's price
:type price: float
"""
cursor = db_connection.cursor()
cursor.execute(
"insert into product(seller_id, name, price_pc) values (%s, %s, %s)",
(product.seller_id, product.name, round(product.price, 2)),
)
db_connection.commit()
def delete_product(product: Product):
cursor = db_connection.cursor()
cursor.execute(
"delete from product where id = %s",
(product.product_id,),
)
db_connection.commit()

View File

@ -1,79 +0,0 @@
from typing import Optional
from app.extensions import db_connection
from app.models.user_model import User
def fetch_by_username(username: str) -> Optional[User]:
cursor = db_connection.cursor(dictionary=True)
cursor.execute("select * from user where username = %s", (username,))
result = cursor.fetchone()
result_user = (
User(
user_id=result["id"],
username=result["username"],
displayname=result["displayname"],
email=result["email"],
password=result["password"],
role_id=result["role_id"],
creation_date=result["creation_date"],
)
if result
else None
)
return result_user
def fetch_by_id(user_id: int) -> Optional[User]:
cursor = db_connection.cursor(dictionary=True)
cursor.execute("select * from user where id = %s", (user_id,))
result = cursor.fetchone()
result_user = (
User(
user_id=result["id"],
username=result["username"],
displayname=result["displayname"],
email=result["email"],
password=result["password"],
role_id=result["role_id"],
creation_date=result["creation_date"],
)
if result
else None
)
return result_user
def insert_user(new_user: User):
cursor = db_connection.cursor(dictionary=True)
cursor.execute(
"insert into user (username, displayname, email, password) values (%s, %s, %s, %s)",
(new_user.username, new_user.displayname, new_user.email, new_user.password),
)
db_connection.commit()
def delete_user(user: User):
cursor = db_connection.cursor(dictionary=True)
cursor.execute("delete from user where id = %s", (user.user_id,))
db_connection.commit()
def update_user(user: User):
cursor = db_connection.cursor(dictionary=True)
cursor.execute(
"update user set username=%s, displayname=%s, email=%s, password=%s where id = %s",
(user.username, user.displayname, user.email, user.password, user.user_id),
)
db_connection.commit()

View File

@ -1,118 +0,0 @@
show_cart_swagger = {
"tags": ["Cart"],
"security": [
{"JWT": []}
],
"responses": {
"200": {
"description": "Current content of user's shopping cart",
"schema": {
"items": {
"count": {"type": "int"},
"date_added": {"type": "string"},
"name": {"type": "string"},
"price_subtotal": {"type": "string"}
},
"example": [
{
"count": 5,
"date_added": "Fri, 08 Mar 2024 08:43:09 GMT",
"name": "Tablet",
"price_subtotal": "1499.95"
},
{
"count": 2,
"date_added": "Fri, 08 Mar 2024 06:43:09 GMT",
"name": "Laptop",
"price_subtotal": "999.95"
}
]
}
}
}
}
add_to_cart_swagger ={
"tags": ["Cart"],
"security": [
{"JWT": []}
],
"parameters": [
{
"name": "product_id",
"description": "ID of product to add to cart.",
"in": "path",
"type": "int",
},
{
"name": "count",
"description": "Count of the products. If not provided, defaults to 1",
"in": "query",
"type": "int",
"default": 1,
"minimum": 1,
"required": False
}
],
"responses": {
"200": {"description": "Successfully added a product to cart"},
"400": {"description": "Causes:\n- Count is < 1"}
}
}
remove_from_cart_swagger = {
"tags": ["Cart"],
"security": [{"JWT": []}],
"parameters": [
{
"name": "product_id",
"in": "path",
"type": "integer",
"description": "ID of the product to be removed from the cart",
"required": True
}
],
"responses": {
"200": {"description": "Successfully removed item from the cart"},
"400": {"description": "Bad Request - Invalid input"},
"500": {"description": "Internal Server Error"}
}
}
update_count_in_cart_swagger = {
"tags": ["Cart"],
"security": [{"JWT": []}],
"description": "Updates the count of products in the user's cart. If the count is less than or equal to 0, the product will be removed from the cart.",
"parameters": [
{
"name": "product_id",
"in": "path",
"type": "integer",
"description": "ID of the product to update in the cart",
"required": True
},
{
"name": "count",
"in": "query",
"type": "integer",
"description": "New count of the product in the cart",
"required": True
}
],
"responses": {
"200": {"description": "Successfully updated item count in the cart"},
"400": {"description": "Bad Request - Invalid input"},
"500": {"description": "Internal Server Error"}
}
}
purchase_swagger = {
"tags": ["Cart"],
"security": [{"JWT": []}],
"description": "Purchases the contents of the user's cart. This action creates a new purchase, moves items from the cart to the purchase history, and clears the cart.",
"responses": {
"200": {"description": "Successfully completed the purchase"},
"400": {"description": "Bad Request - Invalid input or cart is empty"},
"500": {"description": "Internal Server Error"}
}
}

View File

@ -1,18 +0,0 @@
main_swagger = {
"info": {
"title": "Swag Shop",
"version": "0.1",
"description": "Simple shop API using flask and co.\nFeatures include:\n- Not working\n- Successful registration of users\n- Adding items to cart\n- I don't know",
},
"host": "localhost:1236",
"schemes": "http",
"securityDefinitions": {
"JWT": {
"type": "apiKey",
"scheme": "bearer",
"name": "Authorization",
"in": "header",
"description": "JWT Authorization header using the Bearer scheme.\n*Make sure to prefix the token with **Bearer**!*"
}
}
}

View File

@ -1,73 +0,0 @@
get_products_swagger = {
"methods": ["GET"],
"tags": ["Products"],
"parameters": [
],
"responses":
{
"200":
{
"description": "Get a page of products",
"schema":
{
"type": "object",
"properties": {
"message": {"type": "string", "example": "Hello, Flask!"}
}
}
}
}
}
get_product_info_swagger = {
"tags": ["Products"],
"parameters": [
{
"name": "id",
"in": "path",
"type": "integer",
"description": "ID of the product to fetch information for",
"required": True
},
{
"name": "fields",
"in": "query",
"type": "string",
"description": "Comma-separated list of fields to include in the response",
"required": False
}
],
"responses": {
"200": {"description": "Successfully fetched product information"},
"400": {"description": "Bad Request - Invalid input or product doesn't exist"},
"500": {"description": "Internal Server Error"}
}
}
create_product_swagger = {
"methods": ["POST"],
"tags": ["Products"],
"security": [{"JWT": []}],
"parameters": [
{
"name": "name",
"in": "body",
"type": "string",
"description": "Name for the new product",
"required": True
},
{
"name": "price",
"in": "body",
"type": "float",
"description": "Price of the product",
"required": True
}
],
"responses": {
"200": {"description": "Successfully fetched product information"},
"400": {"description": "Bad Request - Invalid input or missing input"},
"500": {"description": "Internal Server Error"}
}
}

View File

@ -1,18 +0,0 @@
root_swagger = {
"methods": ["GET"],
"responses":
{
"200":
{
"description": "A hello world json",
"schema":
{
"type": "object",
"properties":
{
"message": {"type": "string", "example": "Hello, Flask!"}
}
}
}
}
}

View File

@ -1,116 +0,0 @@
register_swagger = {
"methods": ["POST"],
"tags": ["User"],
"description": "Registers a new user in the app. Also sends a notification to the user via the provided email",
"parameters": [
{
"in": "body",
"name": "body",
"description": 'Username, displayname and password of the new user\n- Username can be only lowercase and up to 64 characters\n- Displayname can contain special characters (. _ -) and lower and upper characters\n- Password must be at least 8 characters long, contain both lower and upper characters, numbers and special characters\n- Email has to be in format "name@domain.tld" and up to 64 characters long in total',
"required": True,
"schema": {
"type": "object",
"properties": {
"username": {"type": "string", "example": "mycoolusername"},
"email": {"type": "string", "example": "mymail@dot.com"},
"displayname": {"type": "string", "example": "MyCoolDisplayName"},
"password": {"type": "string", "example": "My5tr0ngP@55w0rd"},
},
},
}
],
}
login_swagger = {
"methods": ["POST"],
"tags": ["User"],
"description": "Logs in using username and password and returns a JWT token for further authorization of requests.\n**The token is valid for 1 hour**",
"parameters": [
{
"in": "body",
"name": "body",
"description": "Username and password payload",
"required": True,
"schema": {
"type": "object",
"properties": {
"username": {"type": "string", "example": "mycoolusername"},
"password": {"type": "string", "example": "MyStrongPassword123"},
},
},
}
],
"responses": {
"200": {
"description": "Returns a fresh token",
"schema": {
"type": "object",
"properties": {
"token": {
"type": "string",
"example": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJmcmVzaCI6ZmFsc2UsImlhdCI6MTcxMDMyMjkyOCwianRpIjoiZDFhYzQxZDktZjA4NC00MmYzLThlMWUtZWFmZjJiNGU1MDAyIiwidHlwZSI6ImFjY2VzcyIsInN1YiI6MjMwMDEsIm5iZiI6MTcxMDMyMjkyOCwiZXhwIjoxNzEwMzI2NTI4fQ.SW7LAi1j5vDOEIvzeN-sy0eHPP9PFJFkXYY029O35w0",
}
},
},
},
"400": {
"description": "Possible causes:\n- Missing username or password from request.\n- Nonexistent username"
},
"401": {"description": "Password is incorrect"},
},
}
logout_swagger = {
"methods": ["DELETE"],
"tags": ["User"],
"security": [{"JWT": []}],
"description": "Logs out the user via provided JWT token",
"parameters": [],
"responses": {"200": {"description": "User successfully logged out"}},
}
update_swagger = {
"methods": ["PUT"],
"tags": ["User"],
"security": [{"JWT": []}],
"description": "Updates user attributes.",
"parameters": [
{
"in": "body",
"name": "body",
"description": "Attributes to update for the user.",
"required": True,
"schema": {
"type": "object",
"properties": {
"new_username": {"type": "string", "example": "mycoolusername"},
"new_email": {"type": "string", "example": "mymail@dot.com"},
"new_displayname": {
"type": "string",
"example": "MyCoolDisplayName",
},
"new_password": {"type": "string", "example": "My5tr0ngP@55w0rd"},
},
},
}
],
"responses": {
"200": {"description": "User attributes updated successfully."},
"400": {"description": "Bad request. Check the request body for errors."},
"401": {"description": "Unauthorized. User must be logged in."},
"409": {"description": "Conflict. Check the response message for details."},
"500": {
"description": "Internal server error. Contact the system administrator."
},
},
}
delete_swagger = {
"methods": ["DELETE"],
"tags": ["User"],
"security": [{"JWT": []}],
"description": "Deletes a user via JWT token",
"parameters": [],
"responses": {"200": {"description": "User successfully deleted"}},
}

View File

@ -1,13 +0,0 @@
from app.extensions import jwt_redis_blocklist
from . import jwt_manager
from app import app
@jwt_manager.token_in_blocklist_loader
def check_if_token_is_revoked(jwt_header, jwt_payload: dict) -> bool:
jti = jwt_payload["jti"]
token_in_redis = jwt_redis_blocklist.get(jti)
return token_in_redis is not None

View File

@ -1,17 +0,0 @@
from flask_mail import Message
from app import flask_mail
from app.mail.message_content import MessageContent
def send_mail(message: MessageContent, recipient: str):
msg = Message(subject=message.subject, recipients=[recipient], body=message.body)
try:
flask_mail.send(msg)
return True
except Exception as e:
print(f"Failed to send email. Error: {e}")
return False

View File

@ -1,4 +0,0 @@
class MessageContent:
def __init__(self, subject, body):
self.subject = subject
self.body = body

View File

@ -1,15 +0,0 @@
NOT_JSON = {"msg": "Request body must be JSON"}, 400
def UNKNOWN_DATABASE_ERROR(e):
return {"msg": f"An unknown error occurred within the database. {e}"}, 500
def MISSING_FIELDS(fields):
return {"msg": f"Missing required fields: {', '.join(fields)}"}, 400
def NO_FIELD_PROVIDED(possible_fields):
return {
"msg": f"No field was provided. At least one of the following is required: {', '.join(possible_fields)}"
}, 400

View File

@ -1,6 +0,0 @@
PRODUCT_LISTING_CREATED_SUCCESSFULLY = {"msg": "Successfully created a brand new product."}, 201
NOT_OWNER_OF_PRODUCT = {"msg": "You don't own this product, therefore you cannot delete it!"}, 400
UNKNOWN_PRODUCT = {"msg": "The product you tried fetching is not known. Try a different product ID."}, 400
SCROLLED_TOO_FAR = {"msg": "You scrolled too far in the pages. Try going back a little again."}, 400

View File

@ -1,26 +0,0 @@
USER_CREATED_SUCCESSFULLY = {"msg": "User created successfully."}, 201
USER_LOGGED_OUT_SUCCESSFULLY = {"msg": "Successfully logged out"}, 200
USER_DELETED_SUCCESSFULLY = {"msg": "User successfully deleted"}, 200
def USER_ACCOUNT_UPDATED_SUCCESSFULLY(updated_attributes):
return {"msg": f"Successfully updated your accounts {', '.join(updated_attributes)}"}, 200
INVALID_USERNAME_FORMAT = {
"msg": "Username is in incorrect format. It must be between 1 and 64 lowercase characters."
}, 400
INVALID_DISPLAYNAME_FORMAT = {
"msg": "Display name is in incorrect format. It must contain only letters, '.', '-', or '_' and be between 1 and 64 characters."
}, 400
INVALID_EMAIL_FORMAT = {"msg": "Email is in incorrect format."}, 400
INVALID_PASSWORD_FORMAT = {
"msg": "Password is in incorrect format. It must be between 8 and 64 characters and contain at least one uppercase letter, one lowercase letter, one digit, and one special character"
}, 400
EMAIL_ALREADY_IN_USE = {"msg": "Email already in use."}, 409
USERNAME_ALREADY_IN_USE = {"msg": "Username already in use."}, 409
USERNAME_NOT_FOUND = {"msg": "Username not found"}, 400
INCORRECT_PASSWORD = {"msg": "Incorrect password"}, 401
UNKNOWN_ERROR = {"msg": "An unknown error occurred with user"}, 500

View File

@ -1,26 +0,0 @@
from app.mail.message_content import MessageContent
USER_EMAIL_SUCCESSFULLY_REGISTERED = MessageContent(
subject="Successfully registered!",
body="Congratulations! Your account has been successfully created.\nThis mail also serves as a test that the email address is correct",
)
USER_EMAIL_SUCCESSFULLY_LOGGED_IN = MessageContent(
subject="New Login detected!",
body="A new login token has been created",
)
USER_EMAIL_SUCCESSFULLY_LOGGED_OUT = MessageContent(
subject="Successfully logged out",
body="A login has been revoked. No further action is needed.",
)
USER_EMAIL_SUCCESSFULLY_UPDATED_ACCOUNT = MessageContent(
subject="Account updated",
body="Your account has been successfully updated. This also means you have been logged out of everywhere",
)
USER_EMAIL_SUCCESSFULLY_DELETED_ACCOUNT = MessageContent(
subject="Account Deleted!",
body="Your account has been deleted. No further action needed",
)

View File

@ -1,4 +1,5 @@
from sqlalchemy import ForeignKey, Column, Integer, JSON, TIMESTAMP, String, Enum from sqlalchemy import ForeignKey, Column, Integer, JSON, TIMESTAMP, String, Enum
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship from sqlalchemy.orm import relationship
from .base_model import Base from .base_model import Base
@ -16,9 +17,9 @@ class User(Base):
first_name = Column(String(64), nullable=True) first_name = Column(String(64), nullable=True)
last_name = Column(String(64), nullable=True) last_name = Column(String(64), nullable=True)
phone_number = Column(String(15), nullable=True) phone_number = Column(String(15), nullable=True)
created_at = Column(TIMESTAMP, default="CURRENT_TIMESTAMP", nullable=True) created_at = Column(TIMESTAMP, default=func.now, nullable=True)
updated_at = Column(TIMESTAMP, default="CURRENT_TIMESTAMP", nullable=True) updated_at = Column(TIMESTAMP, default=func.now, onupdate=func.now, nullable=True)
last_login = Column(TIMESTAMP, default="CURRENT_TIMESTAMP", nullable=True) last_login = Column(TIMESTAMP, nullable=True)
profile_picture = Column(String(100), nullable=True) profile_picture = Column(String(100), nullable=True)
preferences = Column(JSON, nullable=True) preferences = Column(JSON, nullable=True)

View File

@ -1,13 +0,0 @@
from app.extensions import jwt_redis_blocklist
from . import jwt_manager
from app import app
@jwt_manager.token_in_blocklist_loader
def check_if_token_is_revoked(jwt_header, jwt_payload: dict) -> bool:
jti = jwt_payload["jti"]
token_in_redis = jwt_redis_blocklist.get(jti)
return token_in_redis is not None

View File

@ -1,36 +0,0 @@
annotated-types==0.7.0
anyio==4.6.2.post1
certifi==2024.8.30
click==8.1.7
dnspython==2.7.0
email_validator==2.2.0
exceptiongroup==1.2.2
fastapi==0.115.5
fastapi-cli==0.0.5
h11==0.14.0
httpcore==1.0.7
httptools==0.6.4
httpx==0.28.0
idna==3.10
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==3.0.2
mdurl==0.1.2
passlib==1.7.4
pydantic==2.10.2
pydantic_core==2.27.1
Pygments==2.18.0
PyJWT==2.10.1
python-dotenv==1.0.1
python-multipart==0.0.19
PyYAML==6.0.2
rich==13.9.4
shellingham==1.5.4
sniffio==1.3.1
starlette==0.41.3
typer==0.14.0
typing_extensions==4.12.2
uvicorn==0.32.1
uvloop==0.21.0
watchfiles==1.0.0
websockets==14.1