NQLStore, a simple CRUD store python library for any query launguage (or in short nql)
NQLStore provides an oversimplified API for the mundane things of creating, reading,
updating, and deleting data models that are persisted to any SQL- or NoSQL- database.
In total, all we need are four methods and that is it.
Supported databases include:
-
Relational databases like:
- SQLite
- PostgreSQL
- MySQL
-
NoSQL databases like:
- Redis
- MongoDB
If you like our simple API, you can even easily extend it to support your favourite database technology.
- Python +3.10
- Pydantic +2.0
- SQLModel _(optional) - only required for relational databases
- RedisOM (optional) - only required for redis
- Beanie (optional) - only required for MongoDB
Install NQLStore from pypi, with any of the options: sql, mongo, redis, all.
pip install nqlstoreIn your python modules, define your data models as you would define them with your favourite OM package. The only difference is the package you import them from.
Here are examples of OM packages to substitute.
SQL (use SQLModel models)
# models.py
from nqlstore.sql import Field, SQLModel
class Library(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
address: str
name: str
class Book(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
title: str
library_id: int = Field(default=None, foreign_key="library.id")Redis (use RedisOM models)
Take note that JsonModel, EmbeddedJsonModel require RedisJSON, while queries require RedisSearch to be loaded You need to install redis-stack or load the modules manually
# models.py
from typing import List
from nqlstore.redis import Field, JsonModel, EmbeddedJsonModel
class Book(EmbeddedJsonModel):
title: str = Field(index=True)
class Library(JsonModel):
address: str
name: str = Field(index=True)
books: List[Book]Mongo (use Beanie)
# models.py
from nqlstore.mongo import Document, Indexed
class Library(Document):
address: str
name: str
class Book(Document):
title: Indexed(str)
library_id: strInitialize the store that is to host your models.
Similar to how you imported models from specific packages in nqlstore,
import stores from the appropriately named modules innqlstore.
Here are examples for the different database technologies.
Migrations are outside the scope of this package
# main.py
from nqlstore.sql import SQLStore
from .models import Book, Library
if __name__ == "__main__":
store = SQLStore(uri="sqlite+aiosqlite:///database.db")
store.register([
Library,
Book,
])# main.py
from nqlstore.redis import RedisStore
from .models import Book, Library
if __name__ == "__main__":
store = RedisStore(uri="rediss://username:password@localhost:6379/0")
store.register([
Library,
Book,
])# main.py
from nqlstore.mongo import MongoStore
from .models import Book, Library
if __name__ == "__main__":
store = MongoStore(uri="mongodb://localhost:27017", database="testing")
store.register([
Library,
Book,
])In the rest of you application use the four class methods available on the models.
Filtering follows the MongoDb-style
However, for more complex queries, one can also pass in querying styles native to the type of the database,
alongside the MongoBD-style querying. The two queries would be merged as AND queries.
Or one can simply ignore the MongoDB-style querying and stick to the native querying.
The available querying formats include:
- SQL - SQLModel-style
- Redis - RedisOM-style
- MongoDb - MongoDB-style
Inserting new items in a store, call store.insert() method.
new_libraries = await store.insert(Library, [{}, {}])Finding items in a store, call the store.find() method.
The key-word arguments include:
skip (int)- number of records to ignore at the top of the returned results; default is 0.limit (int | None)- maximum number of records to return; default is None.
The querying format is as described above
libraries = await store.find(
Library, query={"name": {"$eq": "Hairora"}, "address" : {"$ne": "Buhimba"}}
)libraries = await store.find(
Library, Library.name == "Hairora", Library.address != "Buhimba"
)libraries = await store.find(
Library, Library.name == "Hairora", query={"address" : {"$ne": "Buhimba"}}
)libraries = await store.find(
Library, query={"name": {"$eq": "Hairora"}, "address" : {"$ne": "Buhimba"}}
)libraries = await store.find(
Library, (Library.name == "Hairora") & (Library.address != "Buhimba")
)libraries = await store.find(
Library, (Library.name == "Hairora"), query={"address" : {"$ne": "Buhimba"}}
)libraries = await store.find(
Library, {"name": "Hairora", "address": {"$ne": "Buhimba"}}
)Updating items in a store, call the store.update() method.
The method returns the newly updated records.
The filters follow the same style as that used when querying as shown above.
Similarly, updates are different for each type of database technology as alluded to earlier.
libraries = await store.update(
Library,
query={"name": {"$eq": "Hairora"}, "address" : {"$ne": "Buhimba"}},
updates={"name": "Foo"},
)libraries = await store.update(
Library,
Library.name == "Hairora", Library.address != "Buhimba",
updates={"name": "Foo"},
)libraries = await store.update(
Library,
Library.name == "Hairora", query={"address" : {"$ne": "Buhimba"}},
updates={"name": "Foo"},
)libraries = await store.update(
Library,
query={"name": {"$eq": "Hairora"}, "address" : {"$ne": "Buhimba"}},
updates={"name": "Foo"},
)libraries = await store.update(
Library,
(Library.name == "Hairora") & (Library.address != "Buhimba"),
updates={"name": "Foo"},
)libraries = await store.update(
Library,
(Library.name == "Hairora"),
query={"address" : {"$ne": "Buhimba"}},
updates={"name": "Foo"},
)Mongo updates are MongoDB-style update dicts
libraries = await store.update(
Library,
{"name": "Hairora", "address": {"$ne": "Buhimba"}},
updates={"$set": {"name": "Foo"}},
)Deleting items in a store, call the store.delete() method.
The filters follow the same style as that used when reading as shown above.
libraries = await store.delete(
Library, query={"name": {"$eq": "Hairora"}, "address" : {"$ne": "Buhimba"}}
)libraries = await store.delete(
Library, Library.name == "Hairora", Library.address != "Buhimba"
)libraries = await store.delete(
Library, Library.name == "Hairora", query={"address" : {"$ne": "Buhimba"}}
)libraries = await store.delete(
Library, query={"name": {"$eq": "Hairora"}, "address" : {"$ne": "Buhimba"}}
)libraries = await store.delete(
Library, (Library.name == "Hairora") & (Library.address != "Buhimba")
)libraries = await store.delete(
Library, (Library.name == "Hairora"), query={"address" : {"$ne": "Buhimba"}}
)libraries = await store.delete(
Library, {"name": "Hairora", "address": {"$ne": "Buhimba"}}
)Contributions are welcome. The docs have to maintained, the code has to be made cleaner, more idiomatic and faster, and there might be need for someone else to take over this repo in case I move on to other things. It happens!
When you are ready, look at the CONTRIBUTIONS GUIDELINES
Copyright (c) 2025 Martin Ahindura
Licensed under the MIT License
"In that day you will ask in My name. I am not saying that I will ask the Father on your behalf. No, the Father himself loves you because you have loved Me and have believed that I came from God."
-- John 16: 26-27
All glory be to God
