linkml_store.api.stores.mongodb package

Adapter for MongoDB document store.

Handles have the form: mongodb://<host>:<port>/<database>

To use this, you must have the pymongo extra installed.

pip install linkml-store[mongodb]

or

pip install linkml-store[all]
class MongoDBCollection(name, parent=None, metadata=None, **kwargs)[source]

Bases: Collection

Adapter for collections in a MongoDB database.

Note

You should not use or manipulate this class directly. Instead, use the general linkml_store.api.Collection

property mongo_collection: Collection
insert(objs, **kwargs)[source]

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:

index(objs, index_name=None, replace=False, unique=False, **kwargs)[source]

Create indexes on the collection.

Parameters:

  • objs (Union[Dict[str, Any], BaseModel, Type, List[Union[Dict[str, Any], BaseModel, Type]]]) – Field(s) to index.

  • index_name (Optional[str]) – Optional name for the index.

  • replace (bool) – If True, the index will be dropped and recreated.

  • unique (bool) – If True, creates a unique index (default: False).

upsert(objs, filter_fields, update_fields=None, **kwargs)[source]

Upsert one or more documents into the MongoDB collection.

Parameters:

  • objs (Union[Dict[str, Any], BaseModel, Type, List[Union[Dict[str, Any], BaseModel, Type]]]) – The document(s) to insert or update.

  • filter_fields (List[str]) – List of field names to use as the filter for matching existing documents.

  • update_fields (Optional[List[str]]) – List of field names to include in the update. If None, all fields are updated.

query(query, limit=None, offset=None, **kwargs)[source]

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:

Any = typing.Any
Dict

alias of Dict

List

alias of List

Union = typing.Union
query_facets(where=None, facet_columns=None, facet_limit=100, **kwargs)[source]

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[Union[str, Tuple[str, ...]]]]) – A list of column names to get facet counts for.

  • facet_limit

Return type:

Dict[Union[str, Tuple[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.

delete(objs, **kwargs)[source]

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)[source]

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)

group_by(group_by_fields, inlined_field='objects', agg_map=None, where=None, **kwargs)[source]

Group objects in the collection by specified fields using MongoDB’s aggregation pipeline.

This implementation leverages MongoDB’s native aggregation capabilities for efficient grouping.

Parameters:

  • group_by_fields (List[str]) – List of fields to group by

  • inlined_field – Field name to store aggregated objects

  • agg_map (Optional[Dict[str, str]]) – Dictionary mapping aggregation types to fields

  • where (Optional[Dict]) – Filter conditions

  • kwargs – Additional arguments

Return type:

QueryResult

Returns:

Query result containing grouped data

class MongoDBDatabase(handle=None, **kwargs)[source]

Bases: Database

An adapter for MongoDB databases.

The LinkML-Store Database abstraction combines mongodb Client and Database.

collection_class

alias of MongoDBCollection

__init__(handle=None, **kwargs)[source]
property native_client: MongoClient
property native_db: Database
commit(**kwargs)[source]

Commit pending changes to the database.

Parameters:

kwargs

Returns:

close(**kwargs)[source]

Close the database.

Parameters:

kwargs

Returns:

drop(**kwargs)[source]

Drop the database and all collections.

>>> from linkml_store.api.client import Client
>>> client = Client()
>>> path = Path("/tmp/test.db")
>>> path.parent.mkdir(exist_ok=True, parents=True)
>>> db = client.attach_database(f"duckdb:///{path}")
>>> db.store({"persons": [{"id": "P1", "name": "John", "age_in_years": 30}]})
>>> coll = db.get_collection("persons")
>>> coll.find({}).num_rows
1
>>> db.drop()
>>> db = client.attach_database("duckdb:///tmp/test.db", alias="test")
>>> coll = db.get_collection("persons")
>>> coll.find({}).num_rows
0
Parameters:

kwargs – additional arguments

query(query, **kwargs)[source]

Run a query against the database.

Examples

>>> from linkml_store.api.client import Client
>>> from linkml_store.api.queries import Query
>>> client = Client()
>>> db = client.attach_database("duckdb", alias="test", recreate_if_exists=True)
>>> collection = db.create_collection("Person")
>>> collection.insert([{"id": "P1", "name": "John"}, {"id": "P2", "name": "Alice"}])
>>> query = Query(from_table="Person", where_clause={"name": "John"})
>>> result = db.query(query)
>>> len(result.rows)
1
>>> result.rows[0]["id"]
'P1'
type query:

Query

param query:

type kwargs:

param kwargs:

rtype:

QueryResult

return:

init_collections()[source]

Initialize collections.

TODO: Not typically called directly: consider making this private :return:

export_database(location, target_format=None, **kwargs)[source]

Export a database to a file or location.

>>> from linkml_store.api.client import Client
>>> client = Client()
>>> db = client.attach_database("duckdb", alias="test")
>>> db.import_database("tests/input/iris.csv", Format.CSV, collection_name="iris")
>>> db.export_database("/tmp/iris.yaml", Format.YAML)
Parameters:

  • location (str) – location of the file

  • target_format (Union[str, Format, None]) – target format

  • kwargs – additional arguments

import_database(location, source_format=None, **kwargs)[source]

Import a database from a file or location.

Parameters:

  • location (str) – location of the file

  • source_format (Optional[str]) – source format

  • kwargs – additional arguments

Submodules