275 lines
13 KiB
Python
275 lines
13 KiB
Python
# Lucas Mathews - Fontys Student ID: 5023572
|
|
# Banking System Manager File
|
|
|
|
from class_client import Client
|
|
from class_account import Account
|
|
from class_transaction import Transaction
|
|
from flask import jsonify, session, request # Imports the Flask modules
|
|
import hashlib # hashlib for password hashing
|
|
import datetime # datetime for timestamps
|
|
import uuid # uuid for unique identifiers
|
|
|
|
|
|
from database import * # Importing the database connection
|
|
|
|
##############
|
|
### System ###
|
|
##############
|
|
|
|
def timestamp(): # Returns the current timestamp
|
|
return (datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
|
|
|
def password_hash(password:str): # Converts a string to SHA512 hash
|
|
return hashlib.sha512(password.encode()).hexdigest()
|
|
|
|
def generate_uuid(): # Generates a unique identifier for transactions
|
|
return str(uuid.uuid4())
|
|
|
|
def generate_uuid_short(): # Generates a short uuid
|
|
return str(uuid.uuid4())[:8]
|
|
|
|
##############
|
|
### Client ###
|
|
##############
|
|
|
|
def login(client_id:str, password:str): # Logs in a user
|
|
password_hash = password_hash(password)
|
|
for client in session.query(Client).all():
|
|
if client.client_id == client_id and client.hash == password_hash:
|
|
session['client_id'] = client_id
|
|
return jsonify({"message": f"{session['username']} logged in succsessfully."}), 200
|
|
return "Invalid client_id or password.", 401
|
|
|
|
def logout():
|
|
if 'client_id' in session:
|
|
session.pop('client_id', None)
|
|
return jsonify({"message": "Logged out"}), 200
|
|
return jsonify({"message": "Not logged in"}), 404
|
|
|
|
def status():
|
|
if 'client_id' in session:
|
|
return jsonify({"message": f"Logged in as {session['username']}"}), 200
|
|
else:
|
|
return jsonify({"message": "Not logged in"}), 400
|
|
|
|
##############
|
|
### Client ###
|
|
##############
|
|
|
|
def get_client(client_id:str): # Returns a specific client in the database
|
|
client = session.query(Client).filter_by(client_id=client_id).one_or_none()
|
|
for client in session.query(Client).all():
|
|
if client.client_id == client_id:
|
|
return jsonify({"name": client.name, "birthdate": client.birthdate, "opening_timestamp": client.opening_timestamp, "address": client.address, "phone_number": client.phone_number, "email": client.email}), 200
|
|
return jsonify({"error": "Client not found"}), 404
|
|
|
|
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()
|
|
notes = kwargs.get("notes", None)
|
|
new_client = Client(client_id, name, birthdate, timestamp(), address, phone_number, email, password_hash(password), notes, 1, 0, None)
|
|
session.add(new_client)
|
|
session.commit()
|
|
return f"New client has been added: name: {name}, uuid: {client_id} ", 200
|
|
|
|
def delete_client(client_id:str): # Deletes a client from the database
|
|
for client in session.query(Client).all():
|
|
if client.client_id == client_id:
|
|
if client.accounts == None:
|
|
session.delete(client)
|
|
session.commit()
|
|
return f"client_id: {client_id} has been removed.", 200
|
|
else:
|
|
return f"client_id: {client_id} has active accounts and can not be removed.", 400
|
|
return f"client_id: {client_id} is not found.", 404
|
|
|
|
def update_client(client_id:str, **kwargs): # Updates a client in the database
|
|
for client in session.query(Client).all():
|
|
if client.client_id == client_id:
|
|
name = kwargs.get("name", None)
|
|
birthdate = kwargs.get("birthdate", None)
|
|
address = kwargs.get("address", None)
|
|
phone_number = kwargs.get("phone_number", None)
|
|
email = kwargs.get("email", None)
|
|
notes = kwargs.get("notes", None)
|
|
if name:
|
|
client.name = name
|
|
if birthdate:
|
|
client.birthdate = birthdate
|
|
if address:
|
|
client.address = address
|
|
if phone_number:
|
|
client.phone_number = phone_number
|
|
if email:
|
|
client.email = email
|
|
if notes:
|
|
client.notes = notes
|
|
session.commit()
|
|
return f"client_id: {client_id} has been updated.", 299
|
|
return f"Client ID: {client_id} is not found." , 400
|
|
|
|
def change_password(client_id:str, password:str, new_password:str): # Changes the password of a client
|
|
old_hash = password_hash(password)
|
|
new_hash = password_hash(new_password)
|
|
for client in session.query(Client).all():
|
|
if client.client_id == client_id:
|
|
if client.hash == old_hash:
|
|
client.hash = new_hash
|
|
session.commit()
|
|
return "Password changed successfully.", 200
|
|
return "Incorrect old password.", 400
|
|
return f"client_id: {client_id} is not found.", 404
|
|
|
|
|
|
###############
|
|
### Account ###
|
|
###############
|
|
|
|
def get_account(account_id:str): # Returns a specific account in the database
|
|
account = session.query(Account).filter_by(account_id=account_id).one_or_none()
|
|
for account in session.query(Account).all():
|
|
if account.account_id == account_id:
|
|
return jsonify({"client_id": account.client_id, "description": account.description, "account_type": account.account_type, "balance": account.balance, "enabled": account.enabled, "notes": account.notes}), 200
|
|
return jsonify({"error": "Account not found"}), 404
|
|
|
|
def add_account(client_id:str, description:str, account_type:str, **kwargs): # Adds a new account to the database
|
|
account_id = generate_uuid_short()
|
|
notes = kwargs.get("notes", None)
|
|
client_found = None
|
|
# Find the client
|
|
for client in session.query(Client).all():
|
|
if client.client_id == client_id:
|
|
client_found = client
|
|
break
|
|
# Check if client was found
|
|
if client_found is None:
|
|
return f"client_id: {client_id} is not found.", 422
|
|
# Add the new account
|
|
new_account = Account(account_id, client_id, description, timestamp(), account_type, 0, 1, notes, None)
|
|
session.add(new_account)
|
|
session.commit()
|
|
return f"New account has been added: description: {description}, uuid: {account_id} ", 200
|
|
|
|
def delete_account(account_id:str): # Deletes an account from the database
|
|
for account in session.query(Account).all():
|
|
if account.account_id == account_id:
|
|
if account.balance == 0:
|
|
session.delete(account)
|
|
session.commit()
|
|
return f"account_id: {account_id} has been removed.", 200
|
|
else:
|
|
return f"account_id: {account_id} has a balance and can not be removed.", 400
|
|
return f"account_id: {account_id} is not found.", 404
|
|
|
|
def update_account(account_id:str, **kwargs): # Updates an account in the database
|
|
for account in session.query(Account).all():
|
|
if account.account_id == account_id:
|
|
description = kwargs.get("description", None)
|
|
account_type = kwargs.get("account_type", None)
|
|
balance = kwargs.get("balance", None)
|
|
enabled = kwargs.get("enabled", None)
|
|
notes = kwargs.get("notes", None)
|
|
if description:
|
|
account.description = description
|
|
if account_type:
|
|
account.account_type = account_type
|
|
if balance:
|
|
account.balance = balance
|
|
if enabled:
|
|
account.enabled = enabled
|
|
if notes:
|
|
account.notes = notes
|
|
session.commit()
|
|
return f"account_id: {account_id} has been updated.", 200
|
|
return f"account_id: {account_id} is not found.", 400
|
|
|
|
|
|
|
|
|
|
###################
|
|
### Transaction ###
|
|
###################
|
|
|
|
def get_transaction(transaction_id:int): # Returns a specific transaction in the database
|
|
transaction = session.query(Transaction).filter_by(transaction_id=transaction_id).one_or_none()
|
|
for transaction in session.query(Transaction).all():
|
|
if transaction.transaction_id == transaction_id:
|
|
return jsonify({"transaction_type": transaction.transaction_type, "amount": transaction.amount, "timestamp": transaction.timestamp, "description": transaction.description, "account_id": transaction.account_id, "recipient_account_id": transaction.recipient_account_id}), 200
|
|
return jsonify({"error": "Transaction not found"}), 404
|
|
|
|
def add_transaction(amount:int, account_id, recipient_account_id, **kwargs): # Adds a new transaction to the database
|
|
transaction_id = generate_uuid()
|
|
for account in session.query(Account).all():
|
|
if account.account_id == account_id:
|
|
account_from = account
|
|
if account.account_id == recipient_account_id:
|
|
account_dest = account
|
|
# Check if account has enough funds
|
|
if account_from.balance < amount:
|
|
return f"Account ID: {account_id} does not have enough funds to transfer {amount}.", 401
|
|
# Perform the transaction
|
|
account_from.balance -= amount
|
|
account_dest.balance += amount
|
|
transaction_type = "transfer"
|
|
session.commit()
|
|
# Create the transaction record
|
|
description = kwargs.get("description", None)
|
|
new_transaction = Transaction(transaction_id, transaction_type, amount, timestamp(), description, account_id, recipient_account_id)
|
|
session.add(new_transaction)
|
|
session.commit()
|
|
return f"New transaction has been added: description: {description}, uuid: {transaction_id} ", 200
|
|
|
|
def transaction_history(account_id:int): # Returns all transactions for a specific account
|
|
result = session.query(Transaction).filter(Transaction.account_id == account_id)
|
|
return jsonify([{"transaction_id": transaction.transaction_id, "transaction_type": transaction.transaction_type, "amount": transaction.amount, "timestamp": transaction.timestamp, "description": transaction.description, "account_number": transaction.account_id, "recipient_account_number": transaction.recipient_account_id} for transaction in result]), 200
|
|
|
|
#####################
|
|
### Administrator ###
|
|
#####################
|
|
|
|
def get_all_clients(): # Returns all clients in the database
|
|
clients = session.query(Client).all()
|
|
return jsonify([{"client_id": client.client_id, "name": client.name, "birthdate": client.birthdate, "opening_timestamp": client.opening_timestamp, "address": client.address, "phone_number": client.phone_number, "email": client.email} for client in clients])
|
|
|
|
def get_all_accounts(): # Returns all accounts in the database
|
|
accounts = session.query(Account).all()
|
|
return jsonify([{"account_id": account.account_id, "client_id": account.client_id, "description": account.description, "open_timestamp": account.open_timestamp, "account_type": account.account_type, "balance": account.balance, "enabled": account.enabled, "notes": account.notes} for account in accounts])
|
|
|
|
def get_all_transactions(): # Returns all transactions in the database
|
|
transactions = session.query(Transaction).all()
|
|
return jsonify([{"transaction_id": transaction.transaction_id, "transaction_type": transaction.transaction_type, "amount": transaction.amount, "timestamp": transaction.timestamp, "description": transaction.description, "account_id": transaction.account_id, "recipient_account_id": transaction.recipient_account_id} for transaction in transactions])
|
|
|
|
|
|
|
|
def apply_interest(account_id:int, interest_rate:float):
|
|
for account in session.query(Account).filter(Account.account_id == account_id):
|
|
if account.account_id == account_id:
|
|
account.balance += account.balance * interest_rate
|
|
session.commit()
|
|
return f"Interest has been applied to Account ID: {account_id}."
|
|
return f"Account ID: {account_id} is not found."
|
|
|
|
def apply_fee(account_id:int, fee:float):
|
|
for account in session.query(Account).all():
|
|
if account.account_id == account_id:
|
|
account.balance -= fee
|
|
session.commit()
|
|
return f"Fee has been applied to Account ID: {account_id}."
|
|
return f"Account ID: {account_id} is not found."
|
|
|
|
def delete_transaction(transaction_id:int):
|
|
DELETE_TRANSACTION = "DELETE FROM transaction WHERE transaction_id=?"
|
|
from api import session, Transaction
|
|
for transaction in session.query(Transaction).all():
|
|
if transaction.transaction_id == transaction_id:
|
|
input(f"Are you sure you would like permanenty delete transaction ID: {transaction_id}? WARNING: This action can not be reversed. (Y/N) ")
|
|
if input == "Y"or input == "y":
|
|
session.execute(DELETE_TRANSACTION, (transaction_id))
|
|
print(f"Transaction ID: {transaction_id} has been removed.")
|
|
else:
|
|
return f"Transaction ID: {transaction_id} has NOT been removed."
|
|
return
|
|
return f"Transaction ID: {transaction_id} is not found."
|
|
|
|
|
|
|