# This file was created by init_project.py - all source files are bundled below # ==================== # backend/app/__init__.py # ==================== # Firefly Analytics App # ==================== # backend/app/config.py # ==================== """ Firefly III Analytics Backend - Configuration """ import os from pydantic_settings import BaseSettings class Settings(BaseSettings): """Application settings from environment variables""" # Firefly III FIREFLY_URL: str = os.getenv("FIREFLY_URL", "https://firefly.scsimedia.duckdns.org") FIREFLY_API_TOKEN: str = os.getenv("FIREFLY_API_TOKEN", "") # Database DATABASE_URL: str = os.getenv("DATABASE_URL", "sqlite:///./data/app.db") # Sync SYNC_INTERVAL_MINUTES: int = int(os.getenv("SYNC_INTERVAL_MINUTES", "30")) # API API_TIMEOUT: int = 30 class Config: env_file = ".env" settings = Settings() # ==================== # backend/app/models.py # ==================== """ Database models for Firefly III Analytics """ from sqlalchemy import Column, Integer, String, Float, DateTime, Boolean, Text, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship from datetime import datetime Base = declarative_base() class Transaction(Base): """Transaction model""" __tablename__ = "transactions" id = Column(Integer, primary_key=True, index=True) firefly_id = Column(Integer, unique=True, index=True) date = Column(DateTime, index=True) amount = Column(Float) description = Column(String) type = Column(String) # withdrawal, deposit, transfer category_id = Column(Integer, ForeignKey("categories.id"), nullable=True) from_account_id = Column(Integer, ForeignKey("accounts.id"), nullable=True) to_account_id = Column(Integer, ForeignKey("accounts.id"), nullable=True) notes = Column(Text, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) category = relationship("Category", foreign_keys=[category_id], back_populates="transactions") from_account = relationship("Account", foreign_keys=[from_account_id]) to_account = relationship("Account", foreign_keys=[to_account_id]) class Category(Base): """Category model""" __tablename__ = "categories" id = Column(Integer, primary_key=True, index=True) firefly_id = Column(Integer, unique=True, index=True) name = Column(String, unique=True, index=True) type = Column(String) # expense, revenue, etc created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) transactions = relationship("Transaction", foreign_keys=[Transaction.category_id], back_populates="category") class Account(Base): """Account model""" __tablename__ = "accounts" id = Column(Integer, primary_key=True, index=True) firefly_id = Column(Integer, unique=True, index=True) name = Column(String, unique=True, index=True) type = Column(String) # asset, expense, revenue, liability, etc balance = Column(Float) currency_code = Column(String) active = Column(Boolean, default=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class SyncLog(Base): """Track sync operations""" __tablename__ = "sync_logs" id = Column(Integer, primary_key=True, index=True) entity_type = Column(String) # transactions, categories, accounts last_sync = Column(DateTime) status = Column(String) # success, failure message = Column(String, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) # ==================== # backend/app/database.py # ==================== """ Database initialization and connection """ from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from app.config import settings from app.models import Base engine = create_engine( settings.DATABASE_URL, connect_args={"check_same_thread": False} if "sqlite" in settings.DATABASE_URL else {} ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) def init_db(): """Initialize database tables""" Base.metadata.create_all(bind=engine) def get_db() -> Session: """Get database session""" db = SessionLocal() try: yield db finally: db.close()