FASTAPP.py

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
import mysql.connector
from datetime import datetime
import os

app = FastAPI()

# === MySQL Connection ===
def get_connection():
    return mysql.connector.connect(
        host=os.getenv("Host"),
        user=os.getenv("User"),
        password=os.getenv("Pass"),
        database=os.getenv("DB"),
        port=os.getenv("Port")
    )

# === Pydantic Models ===

class UserCreate(BaseModel):
    name: str
    email: str
    password_hash: str

class AccountCreate(BaseModel):
    account_name: str
    account_code: str
    account_type: str  # A, C, L, E, I
    description: Optional[str] = None

class JournalLine(BaseModel):
    account_id: int
    is_debit: bool
    amount: float

class JournalEntryCreate(BaseModel):
    entry_date: str
    description: Optional[str] = ""
    created_by: int
    lines: List[JournalLine]

# === Endpoints ===

@app.get("/")
def root():
    return {"message": "ERP Accounting API is running."}


# -- Users --

@app.post("/users/")
def create_user(user: UserCreate):
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("""
            INSERT INTO users (name, email, password_hash)
            VALUES (%s, %s, %s)
        """, (user.name, user.email, user.password_hash))
        conn.commit()
        return {"message": "User created successfully."}
    except mysql.connector.Error as e:
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

# -- Accounts --

@app.post("/accounts/")
def create_account(account: AccountCreate):
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("""
            INSERT INTO accounts (account_name, account_code, account_type, description)
            VALUES (%s, %s, %s, %s)
        """, (account.account_name, account.account_code, account.account_type, account.description))
        conn.commit()
        return {"message": "Account created successfully."}
    except mysql.connector.Error as e:
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

@app.get("/accounts/")
def list_accounts():
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute("SELECT * FROM accounts")
        return cursor.fetchall()
    finally:
        cursor.close()
        conn.close()

# -- Journal Entries --

@app.post("/journal/")
def create_journal(entry: JournalEntryCreate):
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("""
            INSERT INTO journal_entries (entry_date, description, created_by)
            VALUES (%s, %s, %s)
        """, (entry.entry_date, entry.description, entry.created_by))
        journal_id = cursor.lastrowid

        for line in entry.lines:
            cursor.execute("""
                INSERT INTO journal_lines (journal_id, account_id, is_debit, amount)
                VALUES (%s, %s, %s, %s)
            """, (journal_id, line.account_id, line.is_debit, line.amount))

            # Also insert into ledger
            cursor.execute("""
                INSERT INTO ledger (account_id, journal_id, entry_date, debit_amount, credit_amount)
                VALUES (%s, %s, %s, %s, %s)
            """, (
                line.account_id,
                journal_id,
                entry.entry_date,
                line.amount if line.is_debit else 0,
                line.amount if not line.is_debit else 0
            ))

        conn.commit()
        return {"message": "Journal entry created."}
    except mysql.connector.Error as e:
        conn.rollback()
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

# -- Ledger View --

@app.get("/ledger/")
def get_ledger():
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute("""
            SELECT l.entry_date, a.account_name, a.account_type,
                   l.debit_amount, l.credit_amount
            FROM ledger l
            JOIN accounts a ON l.account_id = a.account_id
            ORDER BY l.entry_date ASC
        """)
        return cursor.fetchall()
    finally:
        cursor.close()
        conn.close()

# -- Trial Balance --

@app.get("/trial-balance/")
def get_trial_balance():
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute("""
            SELECT 
                a.account_name,
                a.account_type,
                SUM(l.debit_amount) AS total_debit,
                SUM(l.credit_amount) AS total_credit
            FROM ledger l
            JOIN accounts a ON l.account_id = a.account_id
            GROUP BY a.account_id
        """)
        return cursor.fetchall()
    finally:
        cursor.close()
        conn.close()

# -- Profit & Loss Statement --

@app.get("/report/pnl/")
def get_profit_and_loss():
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute("""
            SELECT 
                a.account_name,
                SUM(l.debit_amount) AS total_debit,
                SUM(l.credit_amount) AS total_credit
            FROM ledger l
            JOIN accounts a ON l.account_id = a.account_id
            WHERE a.account_type IN ('I', 'E')
            GROUP BY a.account_id
        """)
        return cursor.fetchall()
    finally:
        cursor.close()
        conn.close()

# -- Balance Sheet --

@app.get("/report/balance-sheet/")
def get_balance_sheet():
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute("""
            SELECT 
                a.account_name,
                a.account_type,
                SUM(l.debit_amount) AS total_debit,
                SUM(l.credit_amount) AS total_credit
            FROM ledger l
            JOIN accounts a ON l.account_id = a.account_id
            WHERE a.account_type IN ('A', 'L', 'C')
            GROUP BY a.account_id
        """)
        return cursor.fetchall()
    finally:
        cursor.close()
        conn.close()