import mysql.
connector
from mysql.connector import Error
from datetime import date
class FinanceManager:
def __init__(self, host, user, password, database):
try:
self.conn = mysql.connector.connect(
host=host,
user=user,
password=password,
database=database
)
self.cursor = self.conn.cursor(dictionary=True)
print("Connected to database")
except Error as e:
print(f"Error connecting to MySQL: {e}")
def add_account(self, account_name, account_type):
sql = "INSERT INTO accounts (account_name, account_type) VALUES (%s, %s)"
self.cursor.execute(sql, (account_name, account_type))
self.conn.commit()
print(f"Account '{account_name}' added.")
def add_category(self, category_name):
sql = "INSERT INTO categories (category_name) VALUES (%s)"
self.cursor.execute(sql, (category_name,))
self.conn.commit()
print(f"Category '{category_name}' added.")
def add_transaction(self, account_id, category_id, amount, transaction_date,
description):
sql = """
INSERT INTO transactions (account_id, category_id, amount,
transaction_date, description)
VALUES (%s, %s, %s, %s, %s)
"""
self.cursor.execute(sql, (account_id, category_id, amount,
transaction_date, description))
self.conn.commit()
print(f"Transaction added: {amount} on {transaction_date}")
# Update account balance
self.update_account_balance(account_id, amount)
def update_account_balance(self, account_id, amount):
sql = "UPDATE accounts SET balance = balance + %s WHERE account_id = %s"
self.cursor.execute(sql, (amount, account_id))
self.conn.commit()
def get_account_balance(self, account_id):
sql = "SELECT balance FROM accounts WHERE account_id = %s"
self.cursor.execute(sql, (account_id,))
result = self.cursor.fetchone()
if result:
return result['balance']
return None
def list_accounts(self):
self.cursor.execute("SELECT * FROM accounts")
accounts = self.cursor.fetchall()
for acc in accounts:
print(f"ID: {acc['account_id']}, Name: {acc['account_name']}, Type:
{acc['account_type']}, Balance: {acc['balance']}")
def list_transactions(self):
sql = """
SELECT t.transaction_id, a.account_name, c.category_name, t.amount,
t.transaction_date, t.description
FROM transactions t
JOIN accounts a ON t.account_id = a.account_id
JOIN categories c ON t.category_id = c.category_id
ORDER BY t.transaction_date DESC
"""
self.cursor.execute(sql)
transactions = self.cursor.fetchall()
for t in transactions:
print(f"{t['transaction_date']} | {t['account_name']} |
{t['category_name']} | {t['amount']} | {t['description']}")
def close(self):
self.cursor.close()
self.conn.close()
print("Connection closed.")
# Example usage
if __name__ == "__main__":
fm = FinanceManager(host='localhost', user='root', password='yourpassword',
database='company_finance')
# Add accounts and categories (run once)
# fm.add_account('Cash', 'Asset')
# fm.add_account('Sales Revenue', 'Revenue')
# fm.add_category('Sales')
# fm.add_category('Office Supplies')
# Add a transaction
fm.add_transaction(account_id=1, category_id=1, amount=1000.00,
transaction_date=date.today(), description='Initial cash deposit')
# List accounts and transactions
fm.list_accounts()
fm.list_transactions()
fm.close()
