linkml_store.api.stores.chromadb.chromadb_collection module

ChromaDB Collection

class ChromaDBCollection(name, parent=None, metadata=None, **kwargs)

Bases: Collection

A wrapper for ChromaDB collections.

property native_collection: Collection

insert(objs, **kwargs)

Add one or more objects to the collection.

>>> from linkml_store import Client
>>> client = Client()
>>> db = client.attach_database("duckdb", alias="test")
>>> collection = db.create_collection("Person")
>>> objs = [{"id": "P1", "name": "John", "age_in_years": 30}, {"id": "P2", "name": "Alice", "age_in_years": 25}]
>>> collection.insert(objs)

Parameters:

• objs (Union[Dict[str, Any], BaseModel, Type, List[Union[Dict[str, Any], BaseModel, Type]]])

• kwargs

Returns:

delete(objs, **kwargs)

Delete one or more objects from the collection.

First let’s set up a collection:

>>> from linkml_store import Client
>>> client = Client()
>>> db = client.attach_database("duckdb", alias="test")
>>> collection = db.create_collection("Person")
>>> objs = [{"id": "P1", "name": "John", "age_in_years": 30}, {"id": "P2", "name": "Alice", "age_in_years": 25}]
>>> collection.insert(objs)
>>> collection.find({}).num_rows
2

Now let’s delete an object:

>>> collection.delete(objs[0])
>>> collection.find({}).num_rows
1

Deleting the same object again should have no effect:

>>> collection.delete(objs[0])
>>> collection.find({}).num_rows
1

Parameters:

• objs (Union[Dict[str, Any], BaseModel, Type, List[Union[Dict[str, Any], BaseModel, Type]]])

• kwargs

Return type:

int

Returns:

delete_where(where=None, missing_ok=True, **kwargs)

Delete objects that match a query.

First let’s set up a collection:

>>> from linkml_store import Client
>>> client = Client()
>>> db = client.attach_database("duckdb", alias="test")
>>> collection = db.create_collection("Person")
>>> objs = [{"id": "P1", "name": "John", "age_in_years": 30}, {"id": "P2", "name": "Alice", "age_in_years": 25}]
>>> collection.insert(objs)

Now let’s delete an object:

>>> collection.delete_where({"id": "P1"})
>>> collection.find({}).num_rows
1

Match everything:

>>> collection.delete_where({})
>>> collection.find({}).num_rows
0

Parameters:

• where (Optional[Dict[str, Any]]) – where conditions

• missing_ok – if True, do not raise an error if the collection does not exist

• kwargs

Return type:

int

Returns:

number of objects deleted (or -1 if unsupported)

query(query, **kwargs)

Run a query against the collection.

First let’s load a collection:

>>> from linkml_store import Client
>>> from linkml_store.utils.format_utils import load_objects
>>> client = Client()
>>> db = client.attach_database("duckdb")
>>> collection = db.create_collection("Country")
>>> objs = load_objects("tests/input/countries/countries.jsonl")
>>> collection.insert(objs)

Now let’s run a query:

TODO

Parameters:

• query (Query)

• kwargs

Return type:

QueryResult

Returns:

query_facets(where=None, facet_columns=None, facet_limit=100, **kwargs)

Run a query to get facet counts for one or more columns.

This function takes a database connection, a Query object, and a list of column names. It generates and executes a facet count query for each specified column and returns the results as a dictionary where the keys are the column names and the values are pandas DataFrames containing the facet counts.

The facet count query is generated by modifying the original query’s WHERE clause to exclude conditions directly related to the facet column. This allows for counting the occurrences of each unique value in the facet column while still applying the other filtering conditions.

Parameters:

• con – A DuckDB database connection.

• query – A Query object representing the base query.

• facet_columns (Optional[List[str]]) – A list of column names to get facet counts for.

• facet_limit

Return type:

Dict[str, List[Tuple[Any, int]]]

Returns:

A dictionary where keys are column names and values are tuples containing the facet counts for each unique value in the respective column.