INNER CODE UNIT · Python

_execute_export_query

Nuclear-Marmalade/dataforge · forge/dashboard/app.py:613

def _execute_export_query(db, where_clause: Optional[str], limit_clause: str = "") -> list:
    """Execute an export query and return a list of row dicts."""
    query = "SELECT * FROM businesses"
    if where_clause:
        query += f" WHERE {where_clause}"
    query += " ORDER BY created_at DESC"
    if limit_clause:
        query += limit_clause

    with db._backend.connection() as conn:
        if db.is_postgres:
            import psycopg2.extras

            cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
            cur.execute(query)
            rows = [dict(r) for r in cur.fetchall()]
            cur.close()
            return rows

View source record →