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:
- 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 conditionsmissing_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)
- 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
- property native_client: MongoClient
- property native_db: Database
- 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") >>> 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 filetarget_format (
Union
[str
,Format
,None
]) – target formatkwargs – additional arguments
Submodules
- linkml_store.api.stores.mongodb.mongodb_collection module
- linkml_store.api.stores.mongodb.mongodb_database module
MongoDBDatabase
MongoDBDatabase.collection_class
MongoDBDatabase.__init__()
MongoDBDatabase.native_client
MongoDBDatabase.native_db
MongoDBDatabase.commit()
MongoDBDatabase.close()
MongoDBDatabase.drop()
MongoDBDatabase.query()
MongoDBDatabase.init_collections()
MongoDBDatabase.export_database()
MongoDBDatabase.import_database()