mirror of
https://github.com/leonvanzyl/autocoder.git
synced 2026-01-30 22:32:06 +00:00
122f03dc21
- Add CI workflow with Python (ruff lint, security tests) and UI (ESLint, TypeScript, build) jobs - Add ruff, mypy, pytest to requirements.txt - Add pyproject.toml with ruff configuration - Fix import sorting across Python files (ruff --fix) - Fix test_security.py expectations to match actual security policy - Remove invalid 'eof' command from ALLOWED_COMMANDS 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
121 lines
3.6 KiB
Python
121 lines
3.6 KiB
Python
"""
|
|
Database Models and Connection
|
|
==============================
|
|
|
|
SQLite database schema for feature storage using SQLAlchemy.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import Boolean, Column, Integer, String, Text, create_engine
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from sqlalchemy.types import JSON
|
|
|
|
Base = declarative_base()
|
|
|
|
|
|
class Feature(Base):
|
|
"""Feature model representing a test case/feature to implement."""
|
|
|
|
__tablename__ = "features"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
priority = Column(Integer, nullable=False, default=999, index=True)
|
|
category = Column(String(100), nullable=False)
|
|
name = Column(String(255), nullable=False)
|
|
description = Column(Text, nullable=False)
|
|
steps = Column(JSON, nullable=False) # Stored as JSON array
|
|
passes = Column(Boolean, default=False, index=True)
|
|
in_progress = Column(Boolean, default=False, index=True)
|
|
|
|
def to_dict(self) -> dict:
|
|
"""Convert feature to dictionary for JSON serialization."""
|
|
return {
|
|
"id": self.id,
|
|
"priority": self.priority,
|
|
"category": self.category,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"steps": self.steps,
|
|
"passes": self.passes,
|
|
"in_progress": self.in_progress,
|
|
}
|
|
|
|
|
|
def get_database_path(project_dir: Path) -> Path:
|
|
"""Return the path to the SQLite database for a project."""
|
|
return project_dir / "features.db"
|
|
|
|
|
|
def get_database_url(project_dir: Path) -> str:
|
|
"""Return the SQLAlchemy database URL for a project.
|
|
|
|
Uses POSIX-style paths (forward slashes) for cross-platform compatibility.
|
|
"""
|
|
db_path = get_database_path(project_dir)
|
|
return f"sqlite:///{db_path.as_posix()}"
|
|
|
|
|
|
def _migrate_add_in_progress_column(engine) -> None:
|
|
"""Add in_progress column to existing databases that don't have it."""
|
|
from sqlalchemy import text
|
|
|
|
with engine.connect() as conn:
|
|
# Check if column exists
|
|
result = conn.execute(text("PRAGMA table_info(features)"))
|
|
columns = [row[1] for row in result.fetchall()]
|
|
|
|
if "in_progress" not in columns:
|
|
# Add the column with default value
|
|
conn.execute(text("ALTER TABLE features ADD COLUMN in_progress BOOLEAN DEFAULT 0"))
|
|
conn.commit()
|
|
|
|
|
|
def create_database(project_dir: Path) -> tuple:
|
|
"""
|
|
Create database and return engine + session maker.
|
|
|
|
Args:
|
|
project_dir: Directory containing the project
|
|
|
|
Returns:
|
|
Tuple of (engine, SessionLocal)
|
|
"""
|
|
db_url = get_database_url(project_dir)
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# Migrate existing databases to add in_progress column
|
|
_migrate_add_in_progress_column(engine)
|
|
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
return engine, SessionLocal
|
|
|
|
|
|
# Global session maker - will be set when server starts
|
|
_session_maker: Optional[sessionmaker] = None
|
|
|
|
|
|
def set_session_maker(session_maker: sessionmaker) -> None:
|
|
"""Set the global session maker."""
|
|
global _session_maker
|
|
_session_maker = session_maker
|
|
|
|
|
|
def get_db() -> Session:
|
|
"""
|
|
Dependency for FastAPI to get database session.
|
|
|
|
Yields a database session and ensures it's closed after use.
|
|
"""
|
|
if _session_maker is None:
|
|
raise RuntimeError("Database not initialized. Call set_session_maker first.")
|
|
|
|
db = _session_maker()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|