Database Scripts

# scripts/executable/create_stripe_products.py

from sqlalchemy import delete, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from singularity.scripts.base_script import BaseScript
from singularity.common.logger import log as logger


class CreateStripeProductsScript(BaseScript):
    """Create default Stripe product records in the database."""

    def __init__(self):
        super().__init__(
            "create_stripe_products",
            auto_run=True,
            affected_tables=["products", "prices"],
        )

    async def execute(self, db: AsyncSession) -> None:
        logger.info("Creating default Stripe products...")

        await db.execute(
            text("""
                INSERT INTO products (name, stripe_id, active)
                VALUES ('Pro Plan', 'prod_xxx', true)
                ON CONFLICT (stripe_id) DO NOTHING
            """)
        )
        await db.commit()

    async def rollback(self, db: AsyncSession) -> None:
        await db.execute(
            text("DELETE FROM products WHERE stripe_id = 'prod_xxx'")
        )
        await db.commit()

    async def verify(self, db: AsyncSession) -> bool:
        result = await db.execute(
            text("SELECT COUNT(*) FROM products WHERE stripe_id = 'prod_xxx'")
        )
        count = result.scalar()
        return count is not None and count > 0


def main():
    script = CreateStripeProductsScript()
    script.main()

if __name__ == "__main__":
    script = CreateStripeProductsScript()
    script.run_cli()

# scripts/executable/cleanup_orphaned_records.py

from datetime import datetime, timedelta
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from singularity.scripts.base_script import BaseScript
from singularity.common.logger import log as logger


class CleanupOrphanedRecordsScript(BaseScript):
    """Remove temp uploads older than 30 days."""

    def __init__(self):
        super().__init__(
            "cleanup_orphaned_records",
            auto_run=True,
            suppress_errors=True,       # Log failures as warnings, don't block startup
            affected_tables=["temp_uploads"],
        )

    async def execute(self, db: AsyncSession) -> None:
        cutoff = datetime.utcnow() - timedelta(days=30)
        result = await db.execute(
            text("DELETE FROM temp_uploads WHERE created_at < :cutoff"),
            {"cutoff": cutoff},
        )
        await db.commit()
        logger.info(f"Cleaned up {result.rowcount} orphaned records")


def main():
    script = CleanupOrphanedRecordsScript()
    script.main()

if __name__ == "__main__":
    script = CleanupOrphanedRecordsScript()
    script.run_cli()

class OldMigrationScript(BaseScript):
    """Legacy migration -- superseded by v2_migration."""

    def __init__(self):
        super().__init__(
            "old_migration_v1",
            deprecated=True,    # Skipped on startup, rollback still available
        )

    async def execute(self, db: AsyncSession) -> None:
        # Original migration logic (kept for reference)
        ...

    async def rollback(self, db: AsyncSession) -> None:
        # Rollback logic still works via CLI
        ...

from singularity.scripts.runner import run_startup_scripts

# Inside your application lifespan
async def lifespan(app):
    await run_startup_scripts()
    yield

# Create a script with a description
singularity generate script add_user_roles -d "Add default user roles to the database"

# Create and overwrite an existing script
singularity generate script add_user_roles -d "Updated roles" --overwrite

# Run a script by name (default command is "run")
scripts run add_user_roles

# Run with force (re-execute even if already succeeded)
scripts run add_user_roles run --force

# Check execution status
scripts run add_user_roles status

# Rollback a previously executed script
scripts run add_user_roles rollback

# Run a script by its numeric ID (from the list output)
scripts run 3
scripts run 3 status

# List all discovered scripts with detailed info
scripts list

# Show simplified list
scripts list --simple

# Display execution history from the database
scripts history

# Sync content hashes for scripts that predate hash tracking
scripts sync-hashes

# Run all auto-run scripts (same as application startup)
scripts startup