The CSV connector is a file connector that imports claims from local CSV or TSV files.
It uses Python’s built-in csv.DictReader with zero external dependencies.
mapping dictionarydelimiter="\t"csv:{path}:row{num}| Field | Required | Description |
|---|---|---|
subject | Yes | Column name for the claim subject |
predicate | Yes | Column name for the claim predicate |
object | Yes | Column name for the claim object |
subject_type | No | Static value for subject entity type |
predicate_type | No | Static value for predicate type |
object_type | No | Static value for object entity type |
confidence | No | Column name for per-row confidence score |
The CSV connector uses only Python’s standard library csv module.
No additional packages need to be installed.
Ensure your CSV file has a header row. Identify which columns contain the subject, predicate, and object for each claim.
gene,relation,target BRCA1,inhibits,TP53 EGFR,activates,MAPK1
conn = db.connect("csv", path="data/interactions.csv", mapping={"subject": "gene", "predicate": "relation", "object": "target"}, ) result = conn.run(db)
| Parameter | Required | Default | Description |
|---|---|---|---|
path |
Yes | — | Path to the CSV or TSV file |
mapping |
Yes | — | Column-to-claim field mapping dictionary |
delimiter |
No | , |
Field delimiter. Use \t for TSV files. |
confidence |
No | None |
Default confidence for all rows |
source_type |
No | csv_import |
Provenance source type |
save |
No | False |
N/A (no credentials to save) |
conn = db.connect("csv", path="data/interactions.csv", mapping={"subject": "gene", "predicate": "relation", "object": "target"}, ) result = conn.run(db)
conn = db.connect("csv", path="data/edges.tsv", mapping={"subject": "src", "predicate": "rel", "object": "dst"}, delimiter="\t", confidence=0.8, ) result = conn.run(db)
conn = db.connect("csv", path="data/knowledge.csv", mapping={ "subject": "entity_a", "predicate": "relationship", "object": "entity_b", "confidence": "score", "subject_type": "gene", "object_type": "disease", }, ) result = conn.run(db)