Python SDK

A classe LensClient oferece acesso programático a todas as funcionalidades do Lens a partir de código Python. Ela conecta aos buckets registrados, monta views DuckDB e expõe operações de query, IA e cache como chamadas de método.

Import

from dataspoc_lens import LensClient

Uso básico

from dataspoc_lens import LensClient

client = LensClient()

# List tables
tables = client.tables()
print(tables)  # ['customers', 'orders', 'products']

# Run a query
result = client.query("SELECT COUNT(*) as total FROM orders")
print(result["rows"])  # [[48680]]

# Clean up
client.close()

Context manager

Use LensClient como context manager para garantir que a conexão seja fechada automaticamente:

from dataspoc_lens import LensClient

with LensClient() as client:
    tables = client.tables()
    result = client.query("SELECT * FROM orders LIMIT 5")
    print(result["columns"])
    print(result["rows"])

Métodos

tables()

Retorna a lista de nomes de tabelas montadas.

tables = client.tables()
# ['customers', 'orders', 'products']

Retorna: list[str]

schema(table)

Retorna metadados das colunas de uma tabela.

columns = client.schema("orders")
# [
#     {"column_name": "order_id", "data_type": "INTEGER"},
#     {"column_name": "customer_id", "data_type": "INTEGER"},
#     {"column_name": "order_date", "data_type": "DATE"},
#     {"column_name": "total", "data_type": "DOUBLE"},
#     {"column_name": "status", "data_type": "VARCHAR"},
# ]

Parâmetros: table (str) — nome da tabela

Retorna: list[dict] — cada dict tem column_name e data_type

query(sql)

Executa uma query SQL e retorna os resultados.

result = client.query("SELECT status, COUNT(*) as cnt FROM orders GROUP BY status")
print(result["columns"])   # ['status', 'cnt']
print(result["rows"])      # [['completed', 32100], ['pending', 8450], ...]
print(result["row_count"]) # 3
print(result["duration"])  # 0.089

Parâmetros: sql (str) — query SQL a executar

Retorna: dict com as chaves:

• columns — lista de nomes de colunas
• rows — lista de listas de linhas
• row_count — número de linhas retornadas
• duration — tempo da query em segundos

ask(question, **kwargs)

Traduz uma pergunta em linguagem natural para SQL, executa e retorna o resultado.

result = client.ask("What are the top 5 customers by spending?")
print(result["sql"])       # 'SELECT c.name, SUM(o.total) ...'
print(result["columns"])   # ['name', 'total_spent']
print(result["rows"])      # [['Alice Smith', 15420.5], ...]
print(result["error"])     # None

Parâmetros:

• question (str) — pergunta em linguagem natural
• **kwargs — repassados para o módulo de IA (ex.: provider, api_key, model, debug)

Retorna: dict com as chaves:

• sql — query SQL gerada
• columns — lista de nomes de colunas
• rows — lista de listas de linhas
• duration — tempo total em segundos
• error — mensagem de erro ou None

cache_status()

Retorna metadados de todas as tabelas cacheadas.

cached = client.cache_status()
# [
#     {
#         "table": "orders",
#         "status": "fresh",
#         "cached_at": "2026-04-15T10:30:00",
#         "size_bytes": 12902400,
#     }
# ]

Retorna: list[dict] — cada dict tem table, status, cached_at, size_bytes

cache_refresh(table)

Força o refresh do cache local de uma tabela.

info = client.cache_refresh("orders")
print(info)
# {"cached_at": "2026-04-15T14:00:00", "size_bytes": 12902400, "file_count": 4}

Parâmetros: table (str) — nome da tabela

Retorna: dict com cached_at, size_bytes, file_count

Levanta: ValueError se a tabela não for encontrada em nenhum bucket registrado.

cache_refresh_stale()

Atualiza todas as tabelas cacheadas que estão stale.

refreshed = client.cache_refresh_stale()
# [{"table": "customers", "cached_at": "...", "size_bytes": 2200000, "file_count": 1}]

Retorna: list[dict] — uma entrada por tabela atualizada

cache_clear(table=None)

Limpa dados cacheados. Se table for informado, apenas aquela tabela é limpa. Caso contrário, todos os caches são removidos.

# Clear specific table
cleared = client.cache_clear("orders")
# ['orders']

# Clear all
cleared = client.cache_clear()
# ['orders', 'customers', 'products']

Parâmetros: table (str | None) — nome da tabela (opcional)

Retorna: list[str] — nomes das tabelas limpas

close()

Fecha a conexão DuckDB subjacente.

client.close()

Chamado automaticamente ao usar o context manager.

Exemplo completo

from dataspoc_lens import LensClient

with LensClient() as client:
    # Explore
    for table in client.tables():
        cols = client.schema(table)
        print(f"{table}: {len(cols)} columns")

    # Query
    result = client.query("""
        SELECT
            c.name,
            COUNT(o.order_id) as orders,
            SUM(o.total) as revenue
        FROM customers c
        JOIN orders o ON c.id = o.customer_id
        GROUP BY c.name
        ORDER BY revenue DESC
        LIMIT 10
    """)

    for row in result["rows"]:
        print(f"{row[0]}: {row[2]:,.2f}")

    # AI
    answer = client.ask("Which product category has the highest return rate?")
    if answer["error"] is None:
        print(f"SQL: {answer['sql']}")
        for row in answer["rows"]:
            print(row)

    # Cache management
    client.cache_refresh("orders")
    status = client.cache_status()
    for entry in status:
        print(f"{entry['table']}: {entry['status']}")