add user endpoint (99041f68) · Commits · Anthony Jacob / api

app/controller/user.py

0 → 100644

+248 −0

Original line number	Diff line number	Diff line
		from flask import Blueprint, request, jsonify, current_app
		from helpers.security import require_auth
		from helpers.limiter import limiter
		from model.user.user import (
		insertUser,
		getUsers,
		getUsersFull,
		getUser,
		deleteUser,
		updateUser,
		)
		from datetime import date, datetime
		from werkzeug.security import generate_password_hash, check_password_hash


		user_bp = Blueprint("user", __name__)


		@user_bp.route("/users", methods=["GET"])
		@require_auth
		@limiter.limit("10/second", override_defaults=False)
		def get_users():
		user_ids = getUsers()

		if user_ids is False:
		return jsonify({"error": "Database error"}), 500

		return jsonify({"users": user_ids})


		@user_bp.route("/users/full", methods=["GET"])
		@require_auth
		@limiter.limit("10/second", override_defaults=False)
		def get_users_full():
		users = getUsersFull()

		if users is False:
		return jsonify({"error": "Database error"}), 500

		return jsonify({"users": users})


		@user_bp.route("/users/<int:id>", methods=["GET"])
		@require_auth
		@limiter.limit("10/second", override_defaults=False)
		def get_user(id):
		try:
		if not id:
		return jsonify({"error": "Missing User id"}), 400
		else:
		result = getUser(id)
		if result == -1:
		return jsonify({"message": "user not found", "id": id}), 404
		elif result:
		return jsonify(result)
		else:
		return jsonify({"error": "Failed to get user"}), 500

		except Exception as e:
		return jsonify({"error": str(e)}), 500


		@user_bp.route("/users", methods=["PUT"])
		@require_auth
		@limiter.limit("10/second", override_defaults=False)
		def create_user():
		try:
		data = request.get_json()
		login = data.get("login")
		is_active = data.get("is_active")
		password = data.get("password")
		force_jwt_reconnect = data.get("force_jwt_reconnect")
		last_login = data.get("last_login")


		if "login" not in data:
		return jsonify({"error": "Missing login"}), 400

		if "password" not in data:
		return jsonify({"error": "Missing password"}), 400
		else:
		password = generate_password_hash(data.get("password"))


		if "is_active" in data and isinstance(data.get("is_active"), bool):
		is_active = data.get("is_active")
		elif "is_active" in data:
		return (
		jsonify(
		{
		"error": "if you provide is_active, it must be a boolean true or false"
		}
		),
		400,
		)

		if "force_jwt_reconnect" in data and isinstance(data.get("force_jwt_reconnect"), bool):
		force_jwt_reconnect = data.get("force_jwt_reconnect")
		elif "force_jwt_reconnect" in data:
		return (
		jsonify(
		{
		"error": "if you provide force_jwt_reconnect, it must be a boolean true or false"
		}
		),
		400,
		)

		if "last_login" in data:
		if data.get("last_login") is None:
		last_login = "NULL"
		else:
		try:
		last_login = datetime.strptime(
		data.get("last_login"), "%Y-%m-%d %H:%M:%S"
		)
		except ValueError:
		return (
		jsonify(
		{
		"error": "last_login must be in YYYY-MM-DD HH:MM:SS format"
		}
		),
		400,
		)


		user_id = insertUser(login, password, is_active, force_jwt_reconnect, last_login)

		if user_id > 0:
		return (
		jsonify({"message": "User inserted", "user_id": user_id}),
		201,
		)
		elif user_id == -1:
		return jsonify({"error": "user already exist"}), 409
		else:
		return jsonify({"error": "Failed to insert user"}), 500

		except Exception as e:
		return jsonify({"error": str(e)}), 500


		@user_bp.route("/users/<int:id>", methods=["POST"])
		@require_auth
		@limiter.limit("10/second", override_defaults=False)
		def update_user(id):
		try:
		data = request.get_json()
		login = data.get("login")
		is_active = data.get("is_active")
		password = data.get("password")
		force_jwt_reconnect = data.get("force_jwt_reconnect")
		last_login = data.get("last_login")


		if "password" in data:
		password = generate_password_hash(data.get("password"))

		if "is_active" in data and isinstance(data.get("is_active"), bool):
		is_active = data.get("is_active")
		elif "is_active" in data:
		return (
		jsonify(
		{
		"error": "if you provide is_active, it must be a boolean true or false"
		}
		),
		400,
		)

		if "force_jwt_reconnect" in data and isinstance(data.get("force_jwt_reconnect"), bool):
		force_jwt_reconnect = data.get("force_jwt_reconnect")
		elif "force_jwt_reconnect" in data:
		return (
		jsonify(
		{
		"error": "if you provide force_jwt_reconnect, it must be a boolean true or false"
		}
		),
		400,
		)

		if "last_login" in data:
		if data.get("last_login") is None:
		last_login = "NULL"
		else:
		try:
		last_login = datetime.strptime(
		data.get("last_login"), "%Y-%m-%d %H:%M:%S"
		)
		except ValueError:
		return (
		jsonify(
		{
		"error": "last_login must be in YYYY-MM-DD HH:MM:SS format"
		}
		),
		400,
		)




		updatedUser = updateUser(id, login, password, is_active, force_jwt_reconnect, last_login)
		print(f"updatedUser: {updatedUser}")
		if updatedUser == -1:
		return jsonify({"error": "the user is not existing"}), 500
		elif updatedUser == -2:
		return (
		jsonify(
		{"error": "user login is already used by another user Id"}
		),
		500,
		)
		elif updatedUser:
		return (
		jsonify({"message": "User updated", "user": updatedUser}),
		200,
		)
		else:
		return jsonify({"error": "Failed to update user"}), 500

		except Exception as e:
		return jsonify({"error": str(e)}), 500


		@user_bp.route("/users/<int:id>", methods=["DELETE"])
		@require_auth
		@limiter.limit("10/second", override_defaults=False)
		def delete_user(id):
		try:
		if not id:
		return jsonify({"error": "Missing User id"}), 400
		else:
		user = getUser(id)
		if user == -1:
		return jsonify({"message": "user not found", "id": id}), 404
		elif user:
		if deleteUser(id):
		return jsonify({"message": "user deleted", "id": id})
		else:
		return jsonify({"error": "Failed to delete user"}), 500
		else:
		return jsonify({"error": "Failed to retrieve user"}), 500

		except Exception as e:
		return jsonify({"error": str(e)}), 500

app/main.py

+3 −1

Original line number	Diff line number	Diff line
		@@ -5,7 +5,7 @@ from redis import Redis
		from flask_jwt_extended import JWTManager
		from app.helpers.db import db_cursor
		from config import Config
		from controller import diploma, healthcheck, auth, language, keyword, experience, about, service, email
		from controller import diploma, healthcheck, auth, language, keyword, experience, about, service, email, user
		from helpers.limiter import limiter
		from model.security import is_jwt_revoked

		@@ -80,6 +80,8 @@ app.register_blueprint(about.about_bp)

		app.register_blueprint(email.email_bp)

		app.register_blueprint(user.user_bp)


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

app/model/user/user.py

0 → 100644

+198 −0

Original line number	Diff line number	Diff line
		from datetime import datetime
		from flask import current_app
		from typing import Any, Literal
		from helpers.db import db_cursor


		def getUsers():

		getUsersQuery = """SELECT id FROM "user" """
		print("getUsersQuery", getUsersQuery)
		try:
		with db_cursor(dict_results=True) as (conn, cur):
		cur.execute(getUsersQuery)
		rows = cur.fetchall()
		return [row["id"] for row in rows] if rows else []

		except Exception as e:
		current_app.logger.error(f"Database error while retrieving users: {e}")
		return False


		def getUsersFull():

		getUsersQuery = """SELECT id, login, password, is_active, force_jwt_reconnect, last_login FROM "user" ORDER BY id"""

		try:
		with db_cursor(dict_results=True) as (conn, cur):
		cur.execute(getUsersQuery)
		rows = cur.fetchall()
		return rows if rows else []

		except Exception as e:
		current_app.logger.error(f"Database error while retrieving users: {e}")
		return False


		def getUser(userId: int) -> Any \| Literal[-1] \| Literal[False]:

		getUserQuery = """SELECT id, login, password, is_active, force_jwt_reconnect, last_login FROM "user" WHERE id = %s"""

		try:
		with db_cursor(dict_results=True) as (conn, cur):
		cur.execute(getUserQuery, (userId,))
		row = cur.fetchone()
		return row if row else -1

		except Exception as e:
		current_app.logger.error(f"Database error while retrieving users: {e}")
		return False


		def getUserByLogin(userLogin: str) -> Any \| Literal[-1] \| Literal[False]:

		getUserQuery = (
		"""SELECT id, login, password, is_active, force_jwt_reconnect, last_login FROM "user" WHERE login = %s"""
		)

		try:
		with db_cursor(dict_results=True) as (conn, cur):
		cur.execute(getUserQuery, (userLogin,))
		row = cur.fetchone()
		return row if row else -1

		except Exception as e:
		current_app.logger.error(f"Database error while retrieving users: {e}")
		return False


		def isLanguagLoginExists(userLogin: str):
		User = getUserByLogin(userLogin)
		if User and User != -1:
		return True

		return False


		def insertUser(userLogin: str, password:str, is_active: bool, force_jwt_reconnect: bool , last_login: datetime ) -> bool \| int:

		getUserQuery = (
		"""SELECT id FROM "user" WHERE UPPER(login) = UPPER(%s)"""
		)

		InsertUserQuery = """INSERT INTO "user" (login, password, is_active, force_jwt_reconnect, last_login)
		VALUES (%s, %s, %s, %s, %s) RETURNING id;"""

		try:
		with db_cursor() as (conn, cur):
		cur.execute(getUserQuery, (userLogin,))
		row = cur.fetchone()
		if row:
		return -1
		else:
		force_jwt_reconnect = False
		if is_active is None:
		is_active = True
		cur.execute(InsertUserQuery, (userLogin, password, is_active, force_jwt_reconnect, last_login))
		user_id = cur.fetchone()[0]
		conn.commit()
		return user_id

		except Exception as e:
		current_app.logger.error(f"Database error while inserting user: {e}")
		return False


		def deleteUser(userId: int) -> bool:

		deleteUsersQuery = """DELETE FROM "user" WHERE id = %s"""

		try:
		with db_cursor() as (conn, cur):
		cur.execute(deleteUsersQuery, (userId,))
		conn.commit()
		return True

		except Exception as e:
		current_app.logger.error(f"Database error while deleting user: {e}")
		return False


		def updateUser(id: int, userLogin: str, password: str , is_active: bool, force_jwt_reconnect: bool, last_login: datetime):
		getUserCodeQuery = (
		"""SELECT id FROM "user" WHERE UPPER(login) = UPPER(%s)"""
		)

		getUserIdQuery = """SELECT id FROM "user" WHERE id = %s"""

		try:
		with db_cursor() as (conn, cur):
		cur.execute(getUserIdQuery, (id,))
		UserIDToUpdate = cur.fetchone()

		if not UserIDToUpdate:
		return -1

		cur.execute(getUserCodeQuery, (userLogin,))
		ExistId = cur.fetchone()

		if ExistId and ExistId[0] != id:
		return -2
		else:
		update_fields = []
		update_values = []

		if userLogin:
		update_fields.append("login = %s")
		update_values.append(userLogin)

		if password:
		update_fields.append("password = %s")
		update_values.append(password)

		if is_active is not None: # Explicitly checking for None
		update_fields.append("is_active = %s")
		update_values.append(is_active)

		if force_jwt_reconnect is not None:
		update_fields.append("force_jwt_reconnect = %s")
		update_values.append(force_jwt_reconnect)

		if last_login is not None:
		update_fields.append("last_login = %s")
		update_values.append(last_login)

		if update_fields:
		update_query = f"""UPDATE "user" SET {', '.join(update_fields)} WHERE id = %s"""
		update_values.append(id)

		cur.execute(update_query, tuple(update_values))
		conn.commit()
		return id # Returning the updated user ID

		return False # No update performed

		except Exception as e:
		current_app.logger.error(f"Database error while updating user: {e}")
		return False


		def updateLastLogin(userLogin: str, last_login: datetime\|None = None) -> bool:


		# when a default value is set in python function definition,
		# it is evaluated only once when the function is defined,
		# not each time the function is called.
		if last_login is None:
		last_login = datetime.now()

		user = getUserByLogin(userLogin)

		return updateUser(
		id=user["id"],
		userLogin=None,
		password=None,
		is_active=None,
		force_jwt_reconnect=None,
		last_login=last_login,
		)