Minor changes and added descriptions to all funcitons

This commit is contained in:
Lucas Mathews
2024-05-28 11:03:44 +02:00
parent 0732b26af3
commit 7b6ebbd0d2
12 changed files with 132 additions and 87 deletions

2
api.py
View File

@@ -15,6 +15,7 @@ from flask_session import Session # Imports the session module
################# #################
def create_app(): def create_app():
"""Creates the API using Connexion."""
app = connexion.FlaskApp(__name__) app = connexion.FlaskApp(__name__)
app.add_api(CONFIG["api_file"]["name"]) app.add_api(CONFIG["api_file"]["name"])
@@ -26,6 +27,7 @@ def create_app():
return app return app
def API(): def API():
"""Runs the API."""
app = create_app() app = create_app()
app.run(host=CONFIG["server"]["listen_ip"], port=CONFIG["server"]["port"], debug=CONFIG["server"]["debug"]) # Runs the API using the configuration file app.run(host=CONFIG["server"]["listen_ip"], port=CONFIG["server"]["port"], debug=CONFIG["server"]["debug"]) # Runs the API using the configuration file

View File

@@ -1,7 +1,6 @@
import tkinter as tk import tkinter as tk
from tkinter import ttk from tkinter import ttk, messagebox
import customtkinter import customtkinter
import json
from config import CONFIG from config import CONFIG
from connection import get_transactions, format_balance, get_account from connection import get_transactions, format_balance, get_account
import sys import sys
@@ -14,23 +13,20 @@ if len(sys.argv) > 3: # Check if the account description is provided as a comma
account_id = sys.argv[1] account_id = sys.argv[1]
account_description = sys.argv[3] account_description = sys.argv[3]
else: else:
print("Error: Account description not provided.") messagebox.showerror("Error", "Account ID and description were not provided by the server.")
sys.exit(1) sys.exit(1)
def populate_transactions_table(transactions_table, account_id): def populate_transactions_table(transactions_table, account_id):
"""Populate the transactions table with data for the given account ID."""
response = get_transactions(account_id) # Fetch the transactions for the account response = get_transactions(account_id) # Fetch the transactions for the account
print(f"Response from get_transactions: {response}") # Print the response
if response is None or 'data' not in response: if response is None or 'data' not in response:
print(f"Error: Unable to fetch transactions for account {account_id}") messagebox.showerror("Error", "Could not fetch transactions for the account.")
return return
transactions = response['data'] transactions = response['data']
if not isinstance(transactions, list): if not isinstance(transactions, list):
print(f"Error: Expected a list of transactions, but got {type(transactions)}") messagebox.showerror("Error", "Data is not formatted as expected.")
return return
transactions.sort(key=lambda x: x['timestamp'], reverse=True) # Sort transactions by timestamp in descending order
# Sort transactions by timestamp in descending order
transactions.sort(key=lambda x: x['timestamp'], reverse=True)
for transaction in transactions: # Insert the transactions into the transactions_table for transaction in transactions: # Insert the transactions into the transactions_table
transactions_table.insert('', 'end', values=( transactions_table.insert('', 'end', values=(
transaction['transaction_id'], transaction['transaction_id'],

View File

@@ -1,14 +1,13 @@
[server] [server]
ip=0.0.0.0 ip = 0.0.0.0
port=81 port = 81
url=http://127.0.0.1:81 url = http://127.0.0.1:81
[preferences] [preferences]
dark_theme=dark dark_theme = dark
# Modes: light, dark theme = dark-blue
theme=dark-blue
# Themes: blue, dark-blue, green
[client] [client]
default_id=31d90aad default_id = 31d90aad
default_password=Happymeal1 default_password = Happymeal1

View File

@@ -8,9 +8,7 @@ import json
############## ##############
def format_balance(balance): def format_balance(balance):
""" """Formats the balance as a currency string with comma separators."""
Formats the balance as a currency string with comma separators.
"""
return f"{balance:,.2f}" return f"{balance:,.2f}"
##################### #####################
@@ -18,9 +16,7 @@ def format_balance(balance):
##################### #####################
def authenticate_client(client_id, client_password): def authenticate_client(client_id, client_password):
""" """Authenticates a client with the given client_id and client_password."""
Authenticates a client with the given client_id and client_password.
"""
try: try:
response = requests.post(CONFIG["server"]["url"] + "/Client/Login", params={'client_id': client_id, 'password': client_password}) response = requests.post(CONFIG["server"]["url"] + "/Client/Login", params={'client_id': client_id, 'password': client_password})
return response return response
@@ -32,9 +28,7 @@ def authenticate_client(client_id, client_password):
return response return response
def logout_client(): def logout_client():
""" """Logs out the current client."""
Logs out the current client.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)
@@ -48,9 +42,7 @@ def logout_client():
return response return response
def get_client(client_id): def get_client(client_id):
""" """Retrieves the client details for the given client_id."""
Retrieves the client details for the given client_id.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)
@@ -62,9 +54,7 @@ def get_client(client_id):
return {'success': False, 'message': "Could not connect to the server. Please try again later."} return {'success': False, 'message': "Could not connect to the server. Please try again later."}
def update_client(client_id, email=None, phone_number=None, address=None): def update_client(client_id, email=None, phone_number=None, address=None):
""" """Updates the client details for the given client_id."""
Updates the client details for the given client_id.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)
@@ -83,9 +73,7 @@ def update_client(client_id, email=None, phone_number=None, address=None):
return {'success': False, 'message': "Could not connect to the server. Please try again later."} return {'success': False, 'message': "Could not connect to the server. Please try again later."}
def get_accounts(client_id): def get_accounts(client_id):
""" """Retrieves the accounts associated with the given client_id."""
Retrieves the accounts associated with the given client_id.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)
@@ -100,9 +88,7 @@ def get_accounts(client_id):
return {'success': False, 'message': "Could not connect to the server. Please try again later."} return {'success': False, 'message': "Could not connect to the server. Please try again later."}
def get_transactions(account_id): def get_transactions(account_id):
""" """Retrieves the transactions for the given account_id."""
Retrieves the transactions for the given account_id.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)
@@ -117,9 +103,7 @@ def get_transactions(account_id):
return {'success': False, 'message': "Could not connect to the server. Please try again later."} return {'success': False, 'message': "Could not connect to the server. Please try again later."}
def get_account(account_id): def get_account(account_id):
""" """Retrieves the account details for the given account_id."""
Retrieves the account details for the given account_id.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)
@@ -135,9 +119,7 @@ def get_account(account_id):
return {'success': False, 'message': "Could not connect to the server. Please try again later."} return {'success': False, 'message': "Could not connect to the server. Please try again later."}
def update_account(account_id, description=None, notes=None): def update_account(account_id, description=None, notes=None):
""" """Updates the account details for the given account_id."""
Updates the account details for the given account_id.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)
@@ -154,9 +136,7 @@ def update_account(account_id, description=None, notes=None):
return {'success': False, 'message': "Could not connect to the server. Please try again later."} return {'success': False, 'message': "Could not connect to the server. Please try again later."}
def new_transaction(account): def new_transaction(account):
""" """Creates a new transaction for the given account."""
Creates a new transaction for the given account.
"""
try: try:
with open('application\\session_data.json', 'r') as f: with open('application\\session_data.json', 'r') as f:
session_data = json.load(f) session_data = json.load(f)

View File

@@ -15,13 +15,18 @@ frame = None
### Functions ### ### Functions ###
################# #################
def go_to_login():
"""Closes the current window and opens the login page."""
root.destroy()
login_page() # Replace with your function or class that opens the login page
def logout(): def logout():
"""Logs out the client and closes the application.""" """Logs out the client and redirects to the login page."""
response = logout_client() response = logout_client()
json_response = response.json() json_response = response.json()
if json_response['success']: if json_response['success']:
messagebox.showinfo("Logout", "You have been logged out.") messagebox.showinfo("Logout", "You have been logged out.")
root.destroy() go_to_login()
else: else:
messagebox.showerror("Logout failed", json_response['message']) messagebox.showerror("Logout failed", json_response['message'])

View File

@@ -6,13 +6,15 @@ import json
import requests import requests
from connection import authenticate_client from connection import authenticate_client
from config import CONFIG from config import CONFIG
import configparser, sys
################# #################
### Functions ### ### Functions ###
################# #################
def login(): def login():
"""Function to handle the login process.""" """Authenticate the client and open the dashboard if successful."""
client_id = entry_username.get() if entry_username.get() else CONFIG["client"]["default_id"] client_id = entry_username.get() if entry_username.get() else CONFIG["client"]["default_id"]
client_password = entry_password.get() if entry_password.get() else CONFIG["client"]["default_password"] client_password = entry_password.get() if entry_password.get() else CONFIG["client"]["default_password"]
try: try:
@@ -32,6 +34,20 @@ def login():
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
messagebox.showerror("Login failed", f"Could not connect to the server. Please try again later. Error: {str(e)}") messagebox.showerror("Login failed", f"Could not connect to the server. Please try again later. Error: {str(e)}")
def change_dark_theme():
"""Change the theme between dark and light."""
config = configparser.ConfigParser()
config.read('application/app.ini')
if 'preferences' in config:
current_theme = config.get('preferences', 'dark_theme')
new_theme = 'light' if current_theme == 'dark' else 'dark'
config.set('preferences', 'dark_theme', new_theme)
with open('application/app.ini', 'w') as configfile:
config.write(configfile)
os.execl(sys.executable, sys.executable, *sys.argv)
else:
messagebox.showerror("Error", "Could not change the theme. Please check the configuration file.")
############## ##############
### Layout ### ### Layout ###
############## ##############
@@ -64,6 +80,10 @@ entry_password.pack(pady=10)
login_button = customtkinter.CTkButton(root, text="Login", command=login) login_button = customtkinter.CTkButton(root, text="Login", command=login)
login_button.pack(pady=15) login_button.pack(pady=15)
# Create and pack the theme change button
theme_button = customtkinter.CTkButton(root, text="Change Theme", command=change_dark_theme)
theme_button.pack(side='bottom', anchor='w')
# Bind the Return key to the login function # Bind the Return key to the login function
root.bind('<Return>', lambda event=None: login()) root.bind('<Return>', lambda event=None: login())

View File

@@ -1 +1 @@
{"session_cookie": {"session": "snGl2pPHJKUmCdGIH4QdeIgGmzaPxjFZsUe30lxaMpk"}, "client_id": "31d90aad"} {"session_cookie": {"session": "G1SQpUTHsed2pPeVe4-1vtqwOXzE5-HMGZKWxvNqzTs"}, "client_id": "31d90aad"}

View File

@@ -19,6 +19,7 @@ class Account(Base):
transactions = relationship("Transaction", foreign_keys='Transaction.account_id', backref="account") transactions = relationship("Transaction", foreign_keys='Transaction.account_id', backref="account")
def __init__(self, account_id, client_id, description, open_timestamp, account_type, balance, enabled, notes, transactions): def __init__(self, account_id, client_id, description, open_timestamp, account_type, balance, enabled, notes, transactions):
"""Initialises the account object."""
self.account_id = account_id self.account_id = account_id
self.client_id = client_id self.client_id = client_id
self.description = description self.description = description
@@ -30,6 +31,7 @@ class Account(Base):
self.transactions = transactions if transactions is not None else [] self.transactions = transactions if transactions is not None else []
def to_dict(self): def to_dict(self):
"""Returns the account as a dictionary."""
return { return {
"account_id": self.account_id, "account_id": self.account_id,
"client_id": self.client_id, "client_id": self.client_id,

View File

@@ -22,6 +22,7 @@ class Client(Base):
accounts = relationship("Account", backref="client") accounts = relationship("Account", backref="client")
def __init__(self, client_id, name, birthdate, opening_timestamp, address, phone_number, email, hash, notes, enabled, administrator, accounts): def __init__(self, client_id, name, birthdate, opening_timestamp, address, phone_number, email, hash, notes, enabled, administrator, accounts):
"""Initialises the client object."""
self.client_id = client_id self.client_id = client_id
self.name = name self.name = name
self.birthdate = birthdate self.birthdate = birthdate
@@ -36,6 +37,7 @@ class Client(Base):
self.accounts = accounts if accounts is not None else [] self.accounts = accounts if accounts is not None else []
def to_dict(self): def to_dict(self):
"""Returns the client as a dictionary."""
return { return {
"client_id": self.client_id, "client_id": self.client_id,
"name": self.name, "name": self.name,

View File

@@ -16,6 +16,7 @@ class Transaction(Base):
recipient_account_id = Column(String, ForeignKey('accounts.account_id')) recipient_account_id = Column(String, ForeignKey('accounts.account_id'))
def __init__(self, transaction_id, transaction_type, amount, timestamp, description, account_id, recipient_account_id = None): def __init__(self, transaction_id, transaction_type, amount, timestamp, description, account_id, recipient_account_id = None):
"""Initialises the Transaction object."""
self.transaction_id = transaction_id self.transaction_id = transaction_id
self.transaction_type = transaction_type self.transaction_type = transaction_type
self.amount = amount self.amount = amount
@@ -25,6 +26,7 @@ class Transaction(Base):
self.recipient_account_id = recipient_account_id self.recipient_account_id = recipient_account_id
def to_dict(self): def to_dict(self):
"""Converts the Transaction object to a dictionary."""
return { return {
"transaction_id": self.transaction_id, "transaction_id": self.transaction_id,
"transaction_type": self.transaction_type, "transaction_type": self.transaction_type,

View File

@@ -13,6 +13,7 @@ subject = "Test Email"
body = "This is a test email." body = "This is a test email."
def send_email(receiver_email, subject, body): def send_email(receiver_email, subject, body):
"""Sends an email to the specified receiver email address."""
sender_email = CONFIG["smtp"]["sender_email"] sender_email = CONFIG["smtp"]["sender_email"]
message = MIMEMultipart() # Create a multipart message and set headers message = MIMEMultipart() # Create a multipart message and set headers

View File

@@ -22,25 +22,31 @@ otps = {} # Temporary dictionary to store OTPs and their creation time
### System ### ### System ###
############## ##############
def timestamp(): # Returns the current timestamp def timestamp():
"""Returns the current timestamp in the format 'YYYY-MM-DD HH:MM:SS'."""
return (datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")) return (datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
def hash_password(password:str): # Converts a string to SHA512 hash def hash_password(password:str):
"""Hashes a password using the SHA-512 algorithm and returns the hexadecimal representation of the hash."""
return hashlib.sha512(password.encode()).hexdigest() return hashlib.sha512(password.encode()).hexdigest()
def generate_uuid(): # Generates a unique identifier for transactions def generate_uuid():
"""Generates a unique identifier using the UUID4 algorithm and returns it as a string."""
return str(uuid.uuid4()) return str(uuid.uuid4())
def generate_uuid_short(): # Generates a short uuid def generate_uuid_short():
"""Generates a short unique identifier using the UUID4 algorithm and returns the first 8 characters as a string."""
return str(uuid.uuid4())[:8] return str(uuid.uuid4())[:8]
def get_email(client_id:str): # Returns the email of a client def get_email(client_id:str):
"""Returns the email of a client given their client_id. If the client is not found, returns None."""
for client in session.query(Client).all(): for client in session.query(Client).all():
if client.client_id == client_id: if client.client_id == client_id:
return client.email return client.email
return None return None
def format_response(success: bool, message: str = '', data: dict = None): # Formats the response for the API so that it is standardised across all functions def format_response(success: bool, message: str = '', data: dict = None):
"""Formats the response for the API so that it is standardised across all functions."""
response = { response = {
'success': success, 'success': success,
'message': message, 'message': message,
@@ -50,22 +56,28 @@ def format_response(success: bool, message: str = '', data: dict = None): # Form
return jsonify(response) return jsonify(response)
def get_current_client(): def get_current_client():
"""Returns the client_id and administrator status of the currently logged in client. If no client is logged in, returns None, None."""
if 'client_id' not in flask_session:
return None, None
client = flask_session['client_id'] client = flask_session['client_id']
client_obj = session.query(Client).filter_by(client_id=client).one_or_none() client_obj = session.query(Client).filter_by(client_id=client).one_or_none()
if client_obj is None: if client_obj is None:
return None, None return None, None
return client_obj.client_id, client_obj.administrator return client_obj.client_id, client_obj.administrator
def verify_otp(client_id:str, otp:int): # Verifies a one time password for a client def verify_otp(client_id:str, otp:int):
"""Verifies a one time password for a client. Returns True if the OTP is correct and False otherwise."""
if client_id in otps and otps[client_id][0] == otp: if client_id in otps and otps[client_id][0] == otp:
return True return True
return False return False
def delete_otp(client_id:str): # Deletes a one time password for a client def delete_otp(client_id:str):
"""Deletes the OTP for a client."""
if client_id in otps: if client_id in otps:
del otps[client_id] del otps[client_id]
def check_expired_otps(): # Checks and deletes expired OTPs def check_expired_otps():
"""Checks for expired OTPs and deletes them. An OTP is considered expired if it is older than 5 minutes."""
current_time = time.time() current_time = time.time()
expired_otps = [client_id for client_id, (otp, creation_time) in otps.items() if current_time - creation_time > 300] # Find OTPs older than 5 minutes expired_otps = [client_id for client_id, (otp, creation_time) in otps.items() if current_time - creation_time > 300] # Find OTPs older than 5 minutes
for client_id in expired_otps: for client_id in expired_otps:
@@ -75,7 +87,8 @@ def check_expired_otps(): # Checks and deletes expired OTPs
### Authentication ### ### Authentication ###
###################### ######################
def login(client_id:str, password:str): # Logs in a user def login(client_id:str, password:str):
"""Logs in a client using their client_id and password. Returns a success message if the login is successful and an error message otherwise."""
password_hash = hash_password(password) password_hash = hash_password(password)
for client in session.query(Client).all(): for client in session.query(Client).all():
if client.client_id == client_id and client.hash == password_hash: if client.client_id == client_id and client.hash == password_hash:
@@ -84,18 +97,21 @@ def login(client_id:str, password:str): # Logs in a user
return format_response(False, "Invalid client_id or password."), 400 return format_response(False, "Invalid client_id or password."), 400
def logout(): def logout():
"""Logs out a client. Returns a success message if the logout is successful and an error message otherwise."""
if 'client_id' in flask_session: if 'client_id' in flask_session:
flask_session.pop('client_id', None) flask_session.pop('client_id', None)
return format_response(True, "Logged out successfully."), 200 return format_response(True, "Logged out successfully."), 200
return format_response(False, "Not logged in."), 400 return format_response(False, "Not logged in."), 400
def status(): def status():
"""Returns the current status of the client."""
if 'client_id' in flask_session: if 'client_id' in flask_session:
return format_response(True, f"Logged in as {flask_session['client_id']}"), 200 return format_response(True, f"Logged in as {flask_session['client_id']}"), 200
else: else:
return format_response(False, "Not logged in."), 400 return format_response(False, "Not logged in."), 400
def login_required(f): def login_required(f):
"""Decorator function to check if a client is logged in before accessing a route."""
@wraps(f) @wraps(f)
def decorated_function(*args, **kwargs): def decorated_function(*args, **kwargs):
if 'client_id' not in flask_session: if 'client_id' not in flask_session:
@@ -104,6 +120,7 @@ def login_required(f):
return decorated_function return decorated_function
def admin_required(f): def admin_required(f):
"""Decorator function to check if a client is an administrator before accessing a route."""
@wraps(f) @wraps(f)
def decorated_function(*args, **kwargs): def decorated_function(*args, **kwargs):
if 'client_id' not in flask_session: if 'client_id' not in flask_session:
@@ -116,7 +133,8 @@ def admin_required(f):
return decorated_function return decorated_function
@login_required @login_required
def generate_otp(client_id:str): # Generates a one time password for a client def generate_otp(client_id:str):
"""Generates a one time password for a client and sends it to their email address. Returns a success message if the OTP is generated and an error message otherwise."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id: if not is_admin and client_id != current_client_id:
return format_response(False, "You can only generate OTPs for your own client account."), 403 return format_response(False, "You can only generate OTPs for your own client account."), 403
@@ -133,7 +151,8 @@ def generate_otp(client_id:str): # Generates a one time password for a client
############## ##############
@login_required @login_required
def get_client(client_id:str): # Returns a specific client in the database def get_client(client_id:str):
"""Returns a specific client in the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id: if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403 return format_response(False, "You can only view your own client information."), 403
@@ -143,7 +162,8 @@ def get_client(client_id:str): # Returns a specific client in the database
return format_response(False, "Client not found."), 404 return format_response(False, "Client not found."), 404
@login_required @login_required
def update_client(client_id:str, **kwargs): # Updates a client in the database def update_client(client_id:str, **kwargs):
"""Updates a client in the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id: if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403 return format_response(False, "You can only view your own client information."), 403
@@ -172,7 +192,8 @@ def update_client(client_id:str, **kwargs): # Updates a client in the database
return format_response(False, "Client not found."), 404 return format_response(False, "Client not found."), 404
@login_required @login_required
def change_password(client_id:str, password:str, new_password:str, otp:int): # Changes the password of a client def change_password(client_id:str, password:str, new_password:str, otp:int):
"""Changes the password for a client in the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id: if not is_admin and client_id != current_client_id:
return format_response(False, "You can only update your own client information."), 403 return format_response(False, "You can only update your own client information."), 403
@@ -192,9 +213,9 @@ def change_password(client_id:str, password:str, new_password:str, otp:int): # C
@login_required @login_required
def get_accounts(client_id: str): def get_accounts(client_id: str):
"""Returns all accounts for a specific client in the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
if current_client_id is None: if current_client_id is None:
# return an appropriate response or raise an exception
raise Exception("No current client found") raise Exception("No current client found")
if not is_admin and client_id != current_client_id: if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403 return format_response(False, "You can only view your own client information."), 403
@@ -206,7 +227,7 @@ def get_accounts(client_id: str):
############### ###############
@login_required @login_required
def get_account(account_id:str): # Returns a specific account in the database def get_account(account_id:str):
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
account_owner = session.query(Account).filter_by(account_id=account_id).one_or_none().client_id account_owner = session.query(Account).filter_by(account_id=account_id).one_or_none().client_id
if not is_admin and account_owner != current_client_id: if not is_admin and account_owner != current_client_id:
@@ -218,7 +239,8 @@ def get_account(account_id:str): # Returns a specific account in the database
return format_response(False, "Account not found."), 404 return format_response(False, "Account not found."), 404
@login_required @login_required
def add_account(client_id:str, description:str, account_type:str, **kwargs): # Adds a new account to the database def add_account(client_id:str, description:str, account_type:str, **kwargs):
"""Adds a new account to the database. If the client is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
if not is_admin and client_id != current_client_id: if not is_admin and client_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403 return format_response(False, "You can only view your own client information."), 403
@@ -237,7 +259,8 @@ def add_account(client_id:str, description:str, account_type:str, **kwargs): # A
return format_response(True, f"New account has been added: account_id: {account_id}"), 200 return format_response(True, f"New account has been added: account_id: {account_id}"), 200
@login_required @login_required
def update_account(account_id:str, **kwargs): # Updates an account in the database def update_account(account_id:str, **kwargs):
"""Updates an account in the database. If the account is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
account_owner = session.query(Account).filter_by(account_id=account_id).one_or_none().client_id account_owner = session.query(Account).filter_by(account_id=account_id).one_or_none().client_id
if not is_admin and account_owner != current_client_id: if not is_admin and account_owner != current_client_id:
@@ -268,7 +291,8 @@ def update_account(account_id:str, **kwargs): # Updates an account in the databa
################### ###################
@login_required @login_required
def get_transaction(transaction_id:int): # Returns a specific transaction in the database def get_transaction(transaction_id:int):
"""Returns a specific transaction in the database. If the transaction is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
transaction = session.query(Transaction).filter_by(transaction_id=transaction_id).one_or_none() transaction = session.query(Transaction).filter_by(transaction_id=transaction_id).one_or_none()
if not transaction: if not transaction:
@@ -280,7 +304,8 @@ def get_transaction(transaction_id:int): # Returns a specific transaction in the
return format_response(True, "", transaction.to_dict()), 200 return format_response(True, "", transaction.to_dict()), 200
@login_required @login_required
def add_transaction(amount:int, account_id, recipient_account_id, **kwargs): # Adds a new transaction to the database def add_transaction(amount:int, account_id, recipient_account_id, **kwargs):
"""Adds a new transaction to the database. If the account is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
if not is_admin and account_id != current_client_id: if not is_admin and account_id != current_client_id:
return format_response(False, "You can only view your own client information."), 403 return format_response(False, "You can only view your own client information."), 403
@@ -290,23 +315,21 @@ def add_transaction(amount:int, account_id, recipient_account_id, **kwargs): # A
account_from = account account_from = account
if account.account_id == recipient_account_id: if account.account_id == recipient_account_id:
account_dest = account account_dest = account
# Check if account has enough funds if account_from.balance < amount: # Check if account has enough funds
if account_from.balance < amount:
return format_response(False, "Insufficient funds."), 400 return format_response(False, "Insufficient funds."), 400
# Perform the transaction account_from.balance -= amount # Perform the transaction
account_from.balance -= amount
account_dest.balance += amount account_dest.balance += amount
transaction_type = "transfer" transaction_type = "transfer"
session.commit() session.commit()
# Create the transaction record description = kwargs.get("description", None) # Create the transaction record
description = kwargs.get("description", None)
new_transaction = Transaction(transaction_id, transaction_type, amount, timestamp(), description, account_id, recipient_account_id) new_transaction = Transaction(transaction_id, transaction_type, amount, timestamp(), description, account_id, recipient_account_id)
session.add(new_transaction) session.add(new_transaction)
session.commit() session.commit()
return format_response(True, f"New transaction has been added: transaction_id: {transaction_id}"), 200 return format_response(True, f"New transaction has been added: transaction_id: {transaction_id}"), 200
@login_required @login_required
def transaction_history(account_id:int): # Returns all transactions for a specific account def transaction_history(account_id:int):
"""Returns all transactions for a specific account in the database. If the account is not found, returns an error message."""
current_client_id, is_admin = get_current_client() current_client_id, is_admin = get_current_client()
account = session.query(Account).filter_by(account_id=account_id).one_or_none() account = session.query(Account).filter_by(account_id=account_id).one_or_none()
if not account: if not account:
@@ -321,10 +344,10 @@ def transaction_history(account_id:int): # Returns all transactions for a specif
##################### #####################
@admin_required @admin_required
def delete_client(client_id:str): # Deletes a client from the database def delete_client(client_id:str):
"""Deletes a client from the database. If the client is not found, returns an error message."""
if client_id == flask_session['client_id']: if client_id == flask_session['client_id']:
return format_response(False, "You cannot delete your own account."), 400 return format_response(False, "You cannot delete your own account."), 400
for client in session.query(Client).all(): for client in session.query(Client).all():
if client.client_id == client_id: if client.client_id == client_id:
if client.accounts == None: if client.accounts == None:
@@ -336,7 +359,8 @@ def delete_client(client_id:str): # Deletes a client from the database
return format_response(False, "Client not found."), 404 return format_response(False, "Client not found."), 404
@admin_required @admin_required
def delete_account(account_id:str): # Deletes an account from the database def delete_account(account_id:str):
"""Deletes an account from the database. If the account is not found, returns an error message."""
for account in session.query(Account).all(): for account in session.query(Account).all():
if account.account_id == account_id: if account.account_id == account_id:
if account.balance == 0: if account.balance == 0:
@@ -348,22 +372,26 @@ def delete_account(account_id:str): # Deletes an account from the database
return format_response(False, "Account not found."), 404 return format_response(False, "Account not found."), 404
@admin_required @admin_required
def get_all_clients(): # Returns all clients in the database def get_all_clients():
"""Returns all clients in the database."""
clients = session.query(Client).all() clients = session.query(Client).all()
return format_response(True, "", [client.to_dict() for client in clients]), 200 return format_response(True, "", [client.to_dict() for client in clients]), 200
@admin_required @admin_required
def get_all_accounts(): # Returns all accounts in the database def get_all_accounts():
"""Returns all accounts in the database."""
accounts = session.query(Account).all() accounts = session.query(Account).all()
return format_response(True, "", [account.to_dict() for account in accounts]), 200 return format_response(True, "", [account.to_dict() for account in accounts]), 200
@admin_required @admin_required
def get_all_transactions(): # Returns all transactions in the database def get_all_transactions():
"""Returns all transactions in the database."""
transactions = session.query(Transaction).all() transactions = session.query(Transaction).all()
return format_response(True, "", [transaction.to_dict() for transaction in transactions]), 200 return format_response(True, "", [transaction.to_dict() for transaction in transactions]), 200
@admin_required @admin_required
def apply_interest(account_id:int, interest_rate:float): def apply_interest(account_id:int, interest_rate:float):
"""Applies interest to an account based on the interest rate. If the account is not found, returns an error message."""
for account in session.query(Account).filter(Account.account_id == account_id): for account in session.query(Account).filter(Account.account_id == account_id):
if account.account_id == account_id: if account.account_id == account_id:
interest = account.balance * interest_rate interest = account.balance * interest_rate
@@ -374,6 +402,7 @@ def apply_interest(account_id:int, interest_rate:float):
@admin_required @admin_required
def apply_fee(account_id:int, fee:float): def apply_fee(account_id:int, fee:float):
"""Applies a fee to an account based on the fee amount. If the account is not found, returns an error message."""
for account in session.query(Account).all(): for account in session.query(Account).all():
if account.account_id == account_id: if account.account_id == account_id:
account.balance -= fee account.balance -= fee
@@ -383,6 +412,7 @@ def apply_fee(account_id:int, fee:float):
@admin_required @admin_required
def delete_transaction(transaction_id:int): def delete_transaction(transaction_id:int):
"""Deletes a transaction from the database. If the transaction is not found, returns an error message."""
for transaction in session.query(Transaction).all(): for transaction in session.query(Transaction).all():
if transaction.transaction_id == transaction_id: if transaction.transaction_id == transaction_id:
session.delete(transaction) session.delete(transaction)
@@ -392,6 +422,7 @@ def delete_transaction(transaction_id:int):
@admin_required @admin_required
def modify_balance(transaction_id:int, amount:int): def modify_balance(transaction_id:int, amount:int):
"""Modifies the amount of a transaction in the database. If the transaction is not found, returns an error message."""
for transaction in session.query(Transaction).all(): for transaction in session.query(Transaction).all():
if transaction.transaction_id == transaction_id: if transaction.transaction_id == transaction_id:
transaction.amount = amount transaction.amount = amount
@@ -401,6 +432,7 @@ def modify_balance(transaction_id:int, amount:int):
@admin_required @admin_required
def test_account_balances(): def test_account_balances():
"""Checks all account balances in the database and returns a list of discrepancies."""
all_transactions = session.query(Transaction).all()# Get all transactions from the database all_transactions = session.query(Transaction).all()# Get all transactions from the database
calculated_balances = {} # Initialize a dictionary to store the calculated balance for each account calculated_balances = {} # Initialize a dictionary to store the calculated balance for each account
for transaction in all_transactions: # Go through each transaction for transaction in all_transactions: # Go through each transaction
@@ -418,7 +450,8 @@ def test_account_balances():
return format_response(True, "", discrepancies), 200 # Return the list of discrepancies return format_response(True, "", discrepancies), 200 # Return the list of discrepancies
@admin_required @admin_required
def add_client(name:str, birthdate:str, address:str, phone_number:str, email:str, password:str, **kwargs): # Adds a new client to the database def add_client(name:str, birthdate:str, address:str, phone_number:str, email:str, password:str, **kwargs):
"""Adds a new client to the database."""
client_id = generate_uuid_short() client_id = generate_uuid_short()
notes = kwargs.get("notes", None) notes = kwargs.get("notes", None)
new_client = Client(client_id, name, birthdate, timestamp(), address, phone_number, email, hash_password(password), notes, 1, 0, None) new_client = Client(client_id, name, birthdate, timestamp(), address, phone_number, email, hash_password(password), notes, 1, 0, None)
@@ -427,6 +460,7 @@ def add_client(name:str, birthdate:str, address:str, phone_number:str, email:str
return format_response(True, f"New client has been added: client_id: {client_id}"), 200 return format_response(True, f"New client has been added: client_id: {client_id}"), 200
def initialise_database(password:str, email:str): def initialise_database(password:str, email:str):
"""Initialises the database with an administrator client if no clients exist."""
existing_clients = session.query(Client).all() # Check if any clients exist in the database existing_clients = session.query(Client).all() # Check if any clients exist in the database
if not existing_clients: # If no clients exist, create an administrator client if not existing_clients: # If no clients exist, create an administrator client
add_client('ADMINISTRATOR', 'ADMINISTRATOR', 'ADMINISTRATOR', 'ADMINISTRATOR', email, password) # Add the administrator client add_client('ADMINISTRATOR', 'ADMINISTRATOR', 'ADMINISTRATOR', 'ADMINISTRATOR', email, password) # Add the administrator client
@@ -439,6 +473,7 @@ def initialise_database(password:str, email:str):
@admin_required @admin_required
def promote_to_admin(client_id:str): def promote_to_admin(client_id:str):
"""Promotes a client to administrator status. If the client is not found, returns an error message."""
for client in session.query(Client).all(): for client in session.query(Client).all():
if client.client_id == client_id: if client.client_id == client_id:
client.administrator = 1 client.administrator = 1
@@ -448,6 +483,7 @@ def promote_to_admin(client_id:str):
@admin_required @admin_required
def demote_from_admin(client_id:str): def demote_from_admin(client_id:str):
"""Demotes a client from administrator status. If the client is not found, returns an error message."""
for client in session.query(Client).all(): for client in session.query(Client).all():
if client.client_id == client_id: if client.client_id == client_id:
client.administrator = 0 client.administrator = 0