SQL Server

SQL

The SQL Server connector is a SQL connector that queries a Microsoft SQL Server database and maps result rows directly to claims via column mapping.

How it works

Uses fetch() which yields claim dicts — the base class run() handles batching
Connects via pymssql.connect(as_dict=True) for named column access
Each row returned by the SQL query becomes one claim via the column mapping
All values are automatically converted to strings
Provenance is tracked with the query as the source identifier

Column mapping

The mapping dict maps claim fields to column names in your query result. Three fields are required: subject, predicate, and object. Optional fields include subject_type, predicate_type, object_type, and confidence.

mapping = {
    "subject": "column_name",      # Required
    "predicate": "column_name",    # Required
    "object": "column_name",       # Required
    "subject_type": "entity",      # Optional (default: "entity")
    "predicate_type": "relates_to",# Optional
    "object_type": "entity",       # Optional
    "confidence": "column_name",   # Optional
}

1 Ensure SQL Server connectivity

Make sure your machine has network access to the SQL Server instance. The connector uses individual connection parameters (server, database, user, password).

2 Install pymssql

$ pip install pymssql

3 Connect

conn = db.connect("mssql",
    server="localhost",
    database="mydb",
    user="sa",
    password="YourStrong!Passw0rd",
    query="SELECT src, rel, dst FROM edges",
    mapping={"subject": "src", "predicate": "rel", "object": "dst"},
)
result = conn.run(db)

Parameter	Required	Default	Description
`server`	Yes	—	SQL Server hostname or IP
`database`	Yes	—	Database name
`user`	Yes	—	Username
`password`	Yes	—	Password
`query`	Yes	—	SQL query to execute
`mapping`	Yes	—	Column-to-claim field mapping dict
`port`	No	`1433`	SQL Server port
`source_type`	No	`database_import`	Provenance source type
`save`	No	`False`	Encrypt and persist credentials

Basic usage

conn = db.connect("mssql",
    server="sql-server.internal",
    database="analytics",
    user="reader",
    password=os.environ["MSSQL_PASSWORD"],
    query="SELECT gene, relation, target FROM interactions",
    mapping={"subject": "gene", "predicate": "relation", "object": "target"},
)
result = conn.run(db)
print(f"Ingested {result.claims_ingested} claims")

With confidence column

conn = db.connect("mssql",
    server="localhost",
    database="mydb",
    user="sa",
    password="pass",
    query="SELECT src, rel, dst, score FROM edges",
    mapping={
        "subject": "src",
        "predicate": "rel",
        "object": "dst",
        "confidence": "score",
    },
)
result = conn.run(db)

With save (persist encrypted credentials)

conn = db.connect("mssql",
    server=os.environ["MSSQL_HOST"],
    database="prod_db",
    user=os.environ["MSSQL_USER"],
    password=os.environ["MSSQL_PASSWORD"],
    query="SELECT * FROM claims_view",
    mapping={"subject": "subj", "predicate": "pred", "object": "obj"},
    save=True,
)
result = conn.run(db)