azure77 0.1.4

Upload, query and manage data on Azure SQL Server from Python

azure77

Upload, query and manage data on Azure SQL Server from Python.

azure77 is an internal data.world replacement for 77 Indicadores. It stores datasets in Azure SQL using schemas to separate clients, provides a CLI for uploads/queries, and includes benchmark tools for performance optimization.

Installation

pip install azure77

Optional dependencies

pip install pymssql    # alternative driver (no ODBC needed)

Quick Start

from azure77 import ConnectionManager

# Connect via .env file
cm = ConnectionManager()

# Or pass credentials directly
cm = ConnectionManager(
    server="your-server.database.windows.net",
    database="your-db",
    user="your-user",
    password="your-pass",
    driver="pymssql",  # or "pyodbc"; default is "auto"
)

# Test connection
result = cm.test_connection()
print(result.success, result.message, f"{result.latency_ms:.0f}ms")

Configuration

.env file

Create a .env file in your project root:

AZURE77_SERVER=server.database.windows.net
AZURE77_DATABASE=dw77
AZURE77_USER=usuario
AZURE77_PASSWORD=senha

Optional variables:

AZURE77_DRIVER=pymssql          # auto (default), pyodbc, or pymssql
AZURE77_ENV_FILE=path/to/.env   # custom env file path

Environment variables

Variable	Required	Default	Description
AZURE77_SERVER	Yes	—	Azure SQL server hostname
AZURE77_DATABASE	Yes	—	Database name
AZURE77_USER	Yes	—	SQL login username
AZURE77_PASSWORD	Yes	—	SQL login password
AZURE77_DRIVER	No	auto	auto, pyodbc, or pymssql
AZURE77_ENV_FILE	No	.env in CWD	Path to env file

ConnectionManager parameters

ConnectionManager(
    server="...",         # str, loaded from config if None
    database="...",       # str, loaded from config if None
    user="...",           # str, loaded from config if None
    password="...",       # str, loaded from config if None
    timeout=10,           # int, connection timeout in seconds
    trust_cert=False,     # bool, trust server certificate
    driver="pymssql",     # str, "auto", "pyodbc", or "pymssql"
    env_file=None,        # str, explicit .env path
)

API Reference

ConnectionManager

Central connection management for Azure SQL.

from azure77 import ConnectionManager

cm = ConnectionManager(server="...", database="...", user="...", password="...")

# Test connection
result = cm.test_connection()
# result.success: bool
# result.server: str
# result.database: str
# result.message: str
# result.latency_ms: float
# result.error_type: str | None ("authentication", "timeout", "network", "driver")
# result.error_detail: str | None

# Get status without network call
status = cm.get_status()
# status.configured: bool
# status.parameters_set: list[str]
# status.parameters_missing: list[str]

# Get raw connection (caller must close)
conn = cm.get_connection()
cursor = conn.cursor()
cursor.execute("SELECT 1")
conn.close()

ClientService

Create and manage clients (each client maps to a SQL schema).

from azure77 import ConnectionManager, ClientService

cm = ConnectionManager(driver="pymssql")
cs = ClientService(cm)

# Create client (auto-generates slug and schema)
client = cs.create_client("ADL Distribuidora")
# client.slug == "adl-distribuidora"
# client.schema_name == "c_adl-distribuidora"

# Create with explicit slug
client = cs.create_client("TMK Engenharia", slug="tmk")

# List clients
clients = cs.list_clients()                # all
clients = cs.list_clients(active_only=True)  # active only

# Get single client
client = cs.get_client("adl-distribuidora")

# Update client
client = cs.update_client("adl-distribuidora", name="ADL Ltda")

# Deactivate client
client = cs.deactivate_client("adl-distribuidora")

# Validate schema health
validation = cs.validate_schema("adl-distribuidora")
# validation.client_exists: bool
# validation.schema_exists: bool
# validation.is_valid: bool
# validation.error_message: str | None

ImportService

Upload CSV, XLSX, and XLS files to Azure SQL.

from azure77 import ConnectionManager, ClientService
from azure77.upload.service import ImportService

cm = ConnectionManager(driver="pymssql")
cs = ClientService(cm)
upload = ImportService(cm, cs)

# Import file (replace mode — drops and recreates table)
result = upload.import_file(
    file_path="vendas.csv",
    client_slug="adl",
    table_name="vendas",
    mode="replace",        # "replace" or "append"
    encoding=None,         # auto-detect, or force "utf-8", "cp1252", etc.
    separator=None,        # auto-detect, or force ";", ",", "\t"
)

# result.table_name: str           (e.g., "[c_adl-distribuidora].[vendas]")
# result.row_count_total: int
# result.row_count_inserted: int
# result.columns: list[ColumnInfo] (normalized_name, sql_type, nullable)
# result.encoding: str
# result.separator: str
# result.duration_seconds: float
# result.phase_timings: dict[str, float]  (connection, load, swap, log, snapshot)
# result.status: str               ("success" or "error")
# result.error_message: str | None

# Append mode (adds rows to existing table)
result = upload.import_file("novas_vendas.csv", "adl", "vendas", mode="append")

# Preview file without database
from azure77.upload.preview import PreviewService
preview = PreviewService()
file_preview = preview.preview("vendas.csv")
# file_preview.headers: list[str]
# file_preview.columns: list[ColumnInfo]
# file_preview.rows: list[list[str]]
# file_preview.row_count: int

Supported file formats

Format	Extension	Library
CSV	.csv	stdlib csv
Excel	.xlsx	openpyxl (read_only)
Excel legacy	.xls	xlrd

Auto-detection

• Encoding: UTF-8, UTF-8-BOM, Latin1, ISO-8859-1, Windows-1252 (via chardet)
• Separator: ,, ;, \t (via csv.Sniffer + column-count consistency)
• Header: first row used as column names

Upload performance

• CSV rows are streamed with bounded memory.
• pymssql uses native bulk copy with adaptive batches of up to 10,000 rows or approximately 4 MB.
• replace loads a staging table and swaps it atomically at commit.
• append writes directly to the destination table.
• Client lookup, import logging, and snapshots reuse the import connection.

Column normalization

All column names are normalized to SQL-safe lowercase identifiers:

Original	Normalized
Data de Emissão	data_de_emissao
Código Cliente	codigo_cliente
CNPJ/CPF	cnpj_cpf
% Comissão	comissao
1º Vencimento	col_1_vencimento
Valor Total (R$)	valor_total_r

Rules: lowercase, remove accents, replace special chars with _, collapse __, strip leading/trailing _, prefix col_ if starts with digit, deduplicate with _2, _3, ...

Type inference

Types are initially inferred from the first 100 rows, then validated in a streaming pass across the complete file. Incompatible late values widen the column to NVARCHAR before the table is created.

Pattern	SQL Type
Pure integers	INTEGER (stored as BIGINT for overflow safety)
Decimals (Brazilian 1.234,56)	DECIMAL(18,2)
Dates DD/MM/YYYY	DATE
Datetimes DD/MM/YYYY HH:MM:SS	DATETIME2
Mixed/unknown	NVARCHAR(MAX)
Leading zeros (CNPJ/CPF)	NVARCHAR(4000)

QueryService

Execute SQL queries with safety validation and history.

from azure77 import ConnectionManager, ClientService, QueryService

cm = ConnectionManager(driver="pymssql")
cs = ClientService(cm)
qs = QueryService(cm, client_service=cs, admin=False)

# Execute query
result = qs.execute("SELECT TOP 100 * FROM vendas", client_slug="adl")
# result.columns: list[str]
# result.rows: list[list]
# result.row_count: int
# result.execution_time_ms: float
# result.success: bool
# result.was_blocked: bool
# result.error_message: str | None

# Admin mode (bypasses safety validation)
qs_admin = QueryService(cm, admin=True)
result = qs_admin.execute("DROP TABLE old_data")

# Save query
saved = qs.save_query("Top Vendas", "SELECT TOP 10 * FROM vendas ORDER BY valor DESC")
# saved.id: int
# saved.name: str
# saved.sql_text: str

# List saved queries
queries = qs.get_saved_queries()

# Get history
history = qs.get_history(client="adl", limit=50)
# history: list[QueryHistoryEntry]

# Export result to CSV
path = qs.export_result(result, "output.csv", format="csv")

# Export result to Excel
path = qs.export_result(result, "output.xlsx", format="xlsx")

Safety validation

By default (admin=False), the following SQL commands are blocked:

DROP, DELETE, TRUNCATE, ALTER, CREATE, INSERT, UPDATE, GRANT, REVOKE, EXEC, EXECUTE, DENY

Only SELECT and WITH are allowed for non-admin users.

Comments and string literals are correctly stripped before validation:

• SELECT 1; -- DROP TABLE x → safe (DROP is in a comment)
• SELECT 'DROP TABLE users' → safe (DROP is in a string literal)

DatasetService

Browse datasets, view metadata, and inspect import history.

from azure77 import ConnectionManager, ClientService, DatasetService

cm = ConnectionManager(driver="pymssql")
cs = ClientService(cm)
ds = DatasetService(cm, cs)

# List all datasets for a client
datasets = ds.list_datasets("adl")
# datasets: list[DatasetSummary]
# each: .table_name, .row_count, .column_count, .size_bytes, .last_modified

# Get full metadata (including column details)
meta = ds.get_dataset_metadata("adl", "vendas")
# meta: DatasetMetadata
# meta.columns: list[ColumnMetadata]
# each column: .column_name, .data_type, .nullable, .max_length, .precision, .scale

# Get columns only
columns = ds.get_columns("adl", "vendas")

# Search across all clients
results = ds.search_datasets("vendas")

# Get client summary (totals)
summary = ds.get_client_summary("adl")
# summary.total_datasets: int
# summary.total_rows: int
# summary.total_size_bytes: int

# Import history
history = ds.get_import_history("adl", table_name="vendas")
# history: list[ImportLogEntry]

# Version history (snapshots)
versions = ds.get_version_history("adl", "vendas")
# versions: list[VersionEvent]

# Schema diff between snapshots
diff = ds.compare_schema("adl", "vendas", snapshot_id_1=1, snapshot_id_2=2)
# diff.columns_added: list[str]
# diff.columns_removed: list[str]
# diff.columns_changed: list[ColumnInfoChange]

BenchmarkService

Measure upload performance across different strategies.

from azure77 import ConnectionManager, ClientService
from azure77.upload.service import ImportService
from azure77.benchmark.service import BenchmarkService

cm = ConnectionManager(driver="pymssql")
cs = ClientService(cm)
upload = ImportService(cm, cs)
bench = BenchmarkService(cm, upload, client_slug="adl")

# Run single file benchmark
report = bench.run_single("bases_teste/vendas.csv", strategy_id="replace", iterations=3)
# report.strategy_results: list[StrategyResult]
# report.rankings: list[StrategyResult] (sorted by speed)

# Compare strategies across files
report = bench.compare("bases_teste/vendas.csv", iterations=3)

# Sweep multiple files
report = bench.sweep(["bases_teste/a.csv", "bases_teste/b.csv"])

# Get optimization suggestions
suggestions = bench.generate_suggestions(report)
# suggestions.overall_recommendation: str
# suggestions.category_recommendations: list[CategorySuggestion]
# suggestions.variance_warnings: list[VarianceWarning]

CLI Commands

# Connection
azure77 connection test
azure77 connection status

# Clients
azure77 client list
azure77 client create "ADL Distribuidora"
azure77 client create "TMK" --slug tmk
azure77 client get adl-distribuidora
azure77 client update adl-distribuidora --name "ADL Ltda"
azure77 client deactivate adl-distribuidora
azure77 client validate adl-distribuidora

# Upload
azure77 upload preview vendas.csv
azure77 upload preview vendas.csv --rows 10
azure77 upload import vendas.csv --client adl --table vendas --mode replace
azure77 upload import vendas.csv --client adl --table vendas --mode append
azure77 upload logs --client adl

# Datasets
azure77 datasets list adl
azure77 datasets info adl vendas
azure77 datasets columns adl vendas
azure77 datasets history adl vendas

# Query
# (use Python API — CLI query not yet implemented)

# Benchmark
azure77 benchmark run vendas.csv --client adl --strategy replace
azure77 benchmark compare vendas.csv --client adl
azure77 benchmark sweep file1.csv file2.csv --client adl
azure77 benchmark history

Architecture

azure77/
├── ConnectionManager          # Azure SQL connection (pyodbc/pymssql)
├── ClientService              # Client/schema CRUD
├── ImportService              # File upload orchestration
│   ├── FileParser             # CSV/XLSX/XLS parsing + encoding detection
│   ├── ColumnNormalizer       # Name normalization (lowercase, accent-free)
│   ├── TypeInferrer           # SQL type inference
│   └── ImportLogger           # Import audit logs
├── QueryService               # SQL execution + safety
│   ├── SafetyValidator        # Blocks dangerous commands
│   ├── HistoryStore           # SQLite local history
│   ├── SavedQueryStore        # SQLite saved queries
│   └── ResultExporter         # CSV/Excel export
├── DatasetService             # Dataset metadata + browsing
├── BenchmarkService           # Performance testing
└── CLI (Click + Rich)         # Command-line interface

Data storage

• Azure SQL: client data, schemas, import logs, dataset snapshots
• SQLite local (~/.azure77/azure77.db): query history, saved queries
• JSON files (.azure77/benchmarks/): benchmark reports

Schema naming

Each client gets a SQL schema with prefix c_:

Client name	Slug	Schema
ADL Distribuidora	adl-distribuidora	c_adl-distribuidora
TMK Engenharia	tmk	c_tmk
Planmetal	planmetal	c_planmetal

Known Limitations

1. pyodbc authentication: Some Azure SQL configurations only work with pymssql, not pyodbc. Use driver="pymssql" if pyodbc fails with "Login failed".

2. Upsert: Not yet implemented (planned for v2). Only replace and append modes are available.

3. Two-pass imports: Imports scan the file once to validate types and a second time to stream rows to Azure SQL. This keeps memory bounded while avoiding late-row type failures.

4. Single-user CLI: SQLite history is local and not shared between users. For shared history, use the import_logs table in Azure SQL.

5. No web UI: This is a Python library + CLI. No browser interface is provided.

Development

# Install in dev mode
pip install -e ".[dev]"

# Run tests
python -m pytest tests/ -v

# Build package
python -m build

# Publish to PyPI
twine upload dist/*

License

MIT - 77 Indicadores