Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
30f0e44097 | ||
|
|
9da851629d | ||
|
|
ebd808f326 | ||
|
|
e20368287d |
@@ -8,31 +8,39 @@ helpers for the common paths, with a raw escape hatch for everything else.
|
|||||||
`requirements.txt`:
|
`requirements.txt`:
|
||||||
|
|
||||||
```
|
```
|
||||||
mongo @ git+ssh://git@git.rethinkstudios.io/rethink-public/mongo.git@v0.1.3
|
mongo @ git+ssh://git@git.rethinkstudios.io/rethink-public/mongo.git@v0.1.7
|
||||||
```
|
```
|
||||||
|
|
||||||
Direct:
|
Direct:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install "mongo @ git+ssh://git@git.rethinkstudios.io/rethink-public/mongo.git@v0.1.3"
|
pip install "mongo @ git+ssh://git@git.rethinkstudios.io/rethink-public/mongo.git@v0.1.7"
|
||||||
```
|
```
|
||||||
|
|
||||||
Requires `motor` and `pymongo` (pulled transitively).
|
Requires `motor` and `pymongo` (pulled transitively).
|
||||||
|
|
||||||
Drop the `@v0.1.3` suffix from the line above to install the latest unpinned.
|
Drop the `@v0.1.7` suffix from the line above to install the latest unpinned.
|
||||||
|
|
||||||
## Usage
|
## Usage
|
||||||
|
|
||||||
**Object (preferred)** — one client per process:
|
**Object (preferred)** — one client per process:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
from mongo import Mongo
|
from mongo import MongoDB
|
||||||
|
|
||||||
db = Mongo(conn_string, database) # attach as bot.db / app.db
|
db = MongoDB(conn_string, database) # attach as bot.db / app.db
|
||||||
|
await db.connect() # optional: ping to fail-early on a bad URI
|
||||||
users = await db.get_documents("users", {"active": True})
|
users = await db.get_documents("users", {"active": True})
|
||||||
db.close() # on shutdown (sync)
|
db.close() # on shutdown (sync)
|
||||||
|
|
||||||
|
# or with guaranteed cleanup:
|
||||||
|
async with MongoDB(conn_string, database) as db:
|
||||||
|
await db.get_documents("users", {"active": True})
|
||||||
```
|
```
|
||||||
|
|
||||||
|
The class is `MongoDB`; **`Mongo` remains a back-compat alias** (`Mongo = MongoDB`), so
|
||||||
|
existing `Mongo(...)` call sites keep working unchanged.
|
||||||
|
|
||||||
**Module proxy (back-compat)** — arm once, then call bare:
|
**Module proxy (back-compat)** — arm once, then call bare:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
@@ -44,10 +52,26 @@ users = await mongo.get_documents("users", {"active": True})
|
|||||||
Both styles share one client. The proxy exists so legacy call sites keep working
|
Both styles share one client. The proxy exists so legacy call sites keep working
|
||||||
after a one-line `init()`; new code should use the object.
|
after a one-line `init()`; new code should use the object.
|
||||||
|
|
||||||
|
## Naming consistency with the datastore trio
|
||||||
|
|
||||||
|
mongo predates the `redis`/`psql`/`mysql` trio; this version makes it surface-consistent
|
||||||
|
**without breaking anything** (all additive, every old name preserved):
|
||||||
|
|
||||||
|
- class is **`MongoDB`** (with `Mongo` kept as an alias)
|
||||||
|
- **`connect()`** + **`async with`** like the trio (motor connects lazily, so `connect()`
|
||||||
|
just pings to validate early)
|
||||||
|
- **`exists()`** aliases `check_document_exists()`; **`delete()`** aliases
|
||||||
|
`delete_document()` — old names still work, the trio-consistent names are now available
|
||||||
|
|
||||||
|
The one deliberate difference that remains: **mongo swallows** (see below) where the trio
|
||||||
|
is **fail-loud**. That's intentional — flipping it would break existing consumers' branch-
|
||||||
|
on-result control flow.
|
||||||
|
|
||||||
## Error contract
|
## Error contract
|
||||||
|
|
||||||
- **Wrapped methods** log-and-swallow exceptions and return a safe default
|
- **Wrapped methods** log-and-swallow exceptions and return a safe default
|
||||||
(`False` / `[]` / `{}` / `0` / `None`). Branch on the result.
|
(`False` / `[]` / `{}` / `0` / `None`). Branch on the result. (`connect()` is the
|
||||||
|
exception — it raises on a bad connection so you can fail-early.)
|
||||||
- **`db.collection(name)`** (or `db[name]`) returns the raw motor collection:
|
- **`db.collection(name)`** (or `db[name]`) returns the raw motor collection:
|
||||||
full driver surface, no swallowing, **raises**. Use it for anything not wrapped
|
full driver surface, no swallowing, **raises**. Use it for anything not wrapped
|
||||||
(`find_one_and_*` beyond what's exposed, change streams, complex bulk ops).
|
(`find_one_and_*` beyond what's exposed, change streams, complex bulk ops).
|
||||||
@@ -63,6 +87,9 @@ are included.
|
|||||||
|
|
||||||
- `from mongo import func` won't see the proxy (resolved at import, before `init`).
|
- `from mongo import func` won't see the proxy (resolved at import, before `init`).
|
||||||
Use `import mongo` then `mongo.func(...)`.
|
Use `import mongo` then `mongo.func(...)`.
|
||||||
|
- `create_collection(collection, index=...)` seeds **one** index (the param is `index`,
|
||||||
|
renamed from `indexes` in v0.1.6). A key-spec like `[("a",1),("b",1)]` builds one
|
||||||
|
compound index, not two — call `create_index` per index for multiples.
|
||||||
- `find_one_and_update` returns the **after** image by default (`return_after=True`).
|
- `find_one_and_update` returns the **after** image by default (`return_after=True`).
|
||||||
- `bulk_write` takes pymongo ops the caller builds:
|
- `bulk_write` takes pymongo ops the caller builds:
|
||||||
```python
|
```python
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "mongo"
|
name = "mongo"
|
||||||
version = "0.1.3"
|
version = "0.1.7"
|
||||||
description = "async mongodb wrapper over motor with a raw escape hatch"
|
description = "async mongodb wrapper over motor with a raw escape hatch"
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
from .mongo import Mongo, init, instance
|
from .mongo import MongoDB, Mongo, init, instance
|
||||||
|
|
||||||
__all__ = ["Mongo", "init", "instance"]
|
__all__ = ["MongoDB", "Mongo", "init", "instance"]
|
||||||
|
|
||||||
|
|
||||||
def __getattr__(name: str):
|
def __getattr__(name: str):
|
||||||
@@ -13,6 +13,6 @@ def __getattr__(name: str):
|
|||||||
can only forward named attributes. on a Mongo instance both `db[name]` and
|
can only forward named attributes. on a Mongo instance both `db[name]` and
|
||||||
`db.collection(name)` work.
|
`db.collection(name)` work.
|
||||||
"""
|
"""
|
||||||
if not name.startswith("_") and hasattr(Mongo, name):
|
if not name.startswith("_") and hasattr(MongoDB, name):
|
||||||
return getattr(instance(), name)
|
return getattr(instance(), name)
|
||||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||||||
|
|||||||
+100
-15
@@ -2,11 +2,19 @@
|
|||||||
async mongodb wrapper over motor
|
async mongodb wrapper over motor
|
||||||
|
|
||||||
object (preferred), one client per process:
|
object (preferred), one client per process:
|
||||||
from mongo import Mongo
|
from mongo import MongoDB
|
||||||
bot.db = Mongo(conn_string, database)
|
bot.db = MongoDB(conn_string, database)
|
||||||
|
await bot.db.connect() # optional: ping to fail-early
|
||||||
await bot.db.get_documents("users", {"active": True})
|
await bot.db.get_documents("users", {"active": True})
|
||||||
bot.db.close() # on shutdown (sync)
|
bot.db.close() # on shutdown (sync)
|
||||||
|
|
||||||
|
# async with (guaranteed cleanup):
|
||||||
|
async with MongoDB(conn_string, database) as db:
|
||||||
|
await db.get_documents("users", {})
|
||||||
|
|
||||||
|
`Mongo` remains a back-compat alias of `MongoDB` — existing `Mongo(...)` call sites keep
|
||||||
|
working unchanged.
|
||||||
|
|
||||||
module proxy (back-compat), arm once then call bare:
|
module proxy (back-compat), arm once then call bare:
|
||||||
import mongo # not `from mongo import ...`
|
import mongo # not `from mongo import ...`
|
||||||
mongo.init(conn_string, database)
|
mongo.init(conn_string, database)
|
||||||
@@ -16,6 +24,15 @@ errors:
|
|||||||
wrapped methods log and swallow, returning a safe default
|
wrapped methods log and swallow, returning a safe default
|
||||||
(False / [] / {} / 0 / None). .collection(name) / db[name] return
|
(False / [] / {} / 0 / None). .collection(name) / db[name] return
|
||||||
the raw motor collection: full driver surface, raises, nothing swallowed.
|
the raw motor collection: full driver surface, raises, nothing swallowed.
|
||||||
|
NOTE: mongo SWALLOWS by design — this is the one deliberate difference from the
|
||||||
|
newer datastore trio (redis/psql/mysql), which is fail-loud. mongo's contract is
|
||||||
|
kept as-is so existing consumers' branch-on-result control flow doesn't break.
|
||||||
|
|
||||||
|
naming consistency with the trio (all additive — old names still work):
|
||||||
|
- class is `MongoDB` (was `Mongo`, kept as alias)
|
||||||
|
- `connect()` / `async with` like the trio (motor connects lazily, so connect()
|
||||||
|
just pings to validate early)
|
||||||
|
- `exists()` aliases `check_document_exists()`; `delete()` aliases `delete_document()`
|
||||||
|
|
||||||
notes:
|
notes:
|
||||||
- the proxy needs `import mongo`; `from mongo import func` resolves before init
|
- the proxy needs `import mongo`; `from mongo import func` resolves before init
|
||||||
@@ -27,18 +44,51 @@ import logging
|
|||||||
from typing import Any, List, Optional
|
from typing import Any, List, Optional
|
||||||
|
|
||||||
from pymongo import ReturnDocument
|
from pymongo import ReturnDocument
|
||||||
|
from pymongo.errors import BulkWriteError
|
||||||
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection, AsyncIOMotorDatabase
|
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection, AsyncIOMotorDatabase
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class Mongo:
|
class MongoDB:
|
||||||
"""async mongodb wrapper; one client per process, attach to bot as bot.db"""
|
"""async mongodb wrapper; one client per process, attach to bot as bot.db"""
|
||||||
|
|
||||||
def __init__(self, connection_string: str, database: str):
|
def __init__(self, connection_string: str, database: str):
|
||||||
|
# motor builds the client eagerly here (no network I/O yet). a syntactically
|
||||||
|
# malformed URI (e.g. "mongodb://") makes the driver raise InvalidURI at
|
||||||
|
# construction, before connect() — a deploy-time value, so it surfaces at startup
|
||||||
|
# either way and there is nothing to clean up.
|
||||||
self._client = AsyncIOMotorClient(connection_string)
|
self._client = AsyncIOMotorClient(connection_string)
|
||||||
self._db = self._client[database]
|
self._db = self._client[database]
|
||||||
|
|
||||||
|
async def connect(self) -> "MongoDB":
|
||||||
|
"""validate the connection with a ping and return self
|
||||||
|
|
||||||
|
motor connects lazily, so this is optional — call it to fail early on a bad
|
||||||
|
URI/credentials rather than on the first real op (parallel to the trio's
|
||||||
|
connect()). raises on a bad connection, unlike the swallowing wrapped methods.
|
||||||
|
|
||||||
|
an instance is single-use: pymongo 4.x `close()` is irreversible, and connect()
|
||||||
|
only pings (it does not rebuild the client). do not reuse an instance after
|
||||||
|
close()/`async with` exit — construct a fresh MongoDB.
|
||||||
|
"""
|
||||||
|
await self._client.admin.command("ping")
|
||||||
|
return self
|
||||||
|
|
||||||
|
async def __aenter__(self) -> "MongoDB":
|
||||||
|
try:
|
||||||
|
return await self.connect()
|
||||||
|
except BaseException:
|
||||||
|
# connect()/ping failing here would otherwise leak the motor client (built
|
||||||
|
# eagerly in __init__ with a background topology monitor + pool) — __aexit__
|
||||||
|
# is not called when __aenter__ raises. close it before propagating so a
|
||||||
|
# retry loop against a flapping server doesn't accumulate live clients.
|
||||||
|
self.close()
|
||||||
|
raise
|
||||||
|
|
||||||
|
async def __aexit__(self, exc_type, exc, tb) -> None:
|
||||||
|
self.close()
|
||||||
|
|
||||||
def __getitem__(self, collection: str) -> AsyncIOMotorCollection:
|
def __getitem__(self, collection: str) -> AsyncIOMotorCollection:
|
||||||
"""raw collection access via subscript: bot.db['users'].aggregate(...)"""
|
"""raw collection access via subscript: bot.db['users'].aggregate(...)"""
|
||||||
return self._db[collection]
|
return self._db[collection]
|
||||||
@@ -68,13 +118,19 @@ class Mongo:
|
|||||||
# -------------------------------------------------------------------------
|
# -------------------------------------------------------------------------
|
||||||
# collection / index management
|
# collection / index management
|
||||||
|
|
||||||
async def create_collection(self, collection: str, indexes=None) -> bool:
|
async def create_collection(self, collection: str, index=None) -> bool:
|
||||||
"""create a collection, optionally seeding an index"""
|
"""create a collection, optionally seeding ONE index
|
||||||
|
|
||||||
|
`index` is a single index key-spec (e.g. `[("a", 1), ("b", 1)]` builds one
|
||||||
|
compound index, NOT two). for multiple indexes call `create_index` per index.
|
||||||
|
renamed from `indexes` in v0.1.6 to reflect that it seeds a single index; pass it
|
||||||
|
positionally, or update a keyword call site from `indexes=` to `index=`.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
await self._db.create_collection(collection)
|
await self._db.create_collection(collection)
|
||||||
if indexes:
|
if index:
|
||||||
await self._db[collection].create_index(indexes)
|
await self._db[collection].create_index(index)
|
||||||
log.info(f"created indexes for {collection}")
|
log.info(f"created index for {collection}")
|
||||||
return True
|
return True
|
||||||
except Exception:
|
except Exception:
|
||||||
log.exception(f"db.create_collection() for {collection}")
|
log.exception(f"db.create_collection() for {collection}")
|
||||||
@@ -130,10 +186,20 @@ class Mongo:
|
|||||||
async def create_documents(
|
async def create_documents(
|
||||||
self, collection: str, documents: List[dict], ordered: bool = True
|
self, collection: str, documents: List[dict], ordered: bool = True
|
||||||
) -> int:
|
) -> int:
|
||||||
"""insert many documents, returning the inserted count"""
|
"""insert many documents, returning the inserted count
|
||||||
|
|
||||||
|
on a partial failure (e.g. ordered=False with one duplicate key among
|
||||||
|
several documents) the driver raises BulkWriteError AFTER inserting the
|
||||||
|
non-failing documents; that count is read from `e.details['nInserted']`
|
||||||
|
so the return value stays honest — a caller retrying the whole batch on a
|
||||||
|
wrongly-reported 0 would re-insert documents that already landed.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
result = await self._db[collection].insert_many(documents, ordered=ordered)
|
result = await self._db[collection].insert_many(documents, ordered=ordered)
|
||||||
return len(result.inserted_ids)
|
return len(result.inserted_ids)
|
||||||
|
except BulkWriteError as e:
|
||||||
|
log.exception(f"db.create_documents() on {collection}")
|
||||||
|
return e.details.get("nInserted", 0)
|
||||||
except Exception:
|
except Exception:
|
||||||
log.exception(f"db.create_documents() on {collection}")
|
log.exception(f"db.create_documents() on {collection}")
|
||||||
return 0
|
return 0
|
||||||
@@ -224,6 +290,10 @@ class Mongo:
|
|||||||
log.exception(f"db.check_document_exists() on {collection}")
|
log.exception(f"db.check_document_exists() on {collection}")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
async def exists(self, collection: str, target: dict) -> bool:
|
||||||
|
"""trio-consistent alias of check_document_exists"""
|
||||||
|
return await self.check_document_exists(collection, target)
|
||||||
|
|
||||||
async def count_value_in_array(self, collection: str, array_field: str, value: str) -> int:
|
async def count_value_in_array(self, collection: str, array_field: str, value: str) -> int:
|
||||||
"""count documents whose array field contains value"""
|
"""count documents whose array field contains value"""
|
||||||
try:
|
try:
|
||||||
@@ -337,12 +407,18 @@ class Mongo:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
async def upsert_document_field(self, collection: str, target: dict, updates: dict) -> bool:
|
async def upsert_document_field(self, collection: str, target: dict, updates: dict) -> bool:
|
||||||
"""$set fields on a single document, creating it if absent"""
|
"""$set fields on a single document, creating it if absent
|
||||||
|
|
||||||
|
returns True when a document matched or was upserted, even if the write
|
||||||
|
changed nothing (a `$set` to the identical value leaves `modified_count=0`);
|
||||||
|
use `matched_count` so an idempotent no-op upsert on an existing doc isn't
|
||||||
|
misread as a failure — mirrors update_document_field/update_document_operator.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
response = await self._db[collection].update_one(
|
response = await self._db[collection].update_one(
|
||||||
target, {"$set": updates}, upsert=True
|
target, {"$set": updates}, upsert=True
|
||||||
)
|
)
|
||||||
return bool(response.modified_count or response.upserted_id)
|
return bool(response.matched_count or response.upserted_id)
|
||||||
except Exception:
|
except Exception:
|
||||||
log.exception(f"db.upsert_document_field() on {collection}")
|
log.exception(f"db.upsert_document_field() on {collection}")
|
||||||
return False
|
return False
|
||||||
@@ -463,6 +539,10 @@ class Mongo:
|
|||||||
log.exception(f"db.delete_document() on {collection}")
|
log.exception(f"db.delete_document() on {collection}")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
async def delete(self, collection: str, target: dict) -> int:
|
||||||
|
"""trio-consistent alias of delete_document (single-document delete)"""
|
||||||
|
return await self.delete_document(collection, target)
|
||||||
|
|
||||||
async def delete_documents(self, collection: str, target: dict) -> int:
|
async def delete_documents(self, collection: str, target: dict) -> int:
|
||||||
"""delete all matching documents, returning deleted count"""
|
"""delete all matching documents, returning deleted count"""
|
||||||
try:
|
try:
|
||||||
@@ -516,6 +596,11 @@ class Mongo:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
# back-compat alias: the class was historically `Mongo`; existing `Mongo(...)` call
|
||||||
|
# sites keep working unchanged
|
||||||
|
Mongo = MongoDB
|
||||||
|
|
||||||
|
|
||||||
# -----------------------------------------------------------------------------
|
# -----------------------------------------------------------------------------
|
||||||
# backwards-compat module proxy
|
# backwards-compat module proxy
|
||||||
# - lets legacy call sites keep using `await mongo.get_documents(...)`
|
# - lets legacy call sites keep using `await mongo.get_documents(...)`
|
||||||
@@ -524,17 +609,17 @@ class Mongo:
|
|||||||
# - does NOT work with `from mongo import func` (resolved at import,
|
# - does NOT work with `from mongo import func` (resolved at import,
|
||||||
# before init runs) — switch those sites to `import mongo`
|
# before init runs) — switch those sites to `import mongo`
|
||||||
|
|
||||||
_default: Optional[Mongo] = None
|
_default: Optional[MongoDB] = None
|
||||||
|
|
||||||
|
|
||||||
def init(connection_string: str, database: str) -> Mongo:
|
def init(connection_string: str, database: str) -> MongoDB:
|
||||||
"""arm the module-level default instance and return it"""
|
"""arm the module-level default instance and return it"""
|
||||||
global _default
|
global _default
|
||||||
_default = Mongo(connection_string, database)
|
_default = MongoDB(connection_string, database)
|
||||||
return _default
|
return _default
|
||||||
|
|
||||||
|
|
||||||
def instance() -> Mongo:
|
def instance() -> MongoDB:
|
||||||
"""return the default instance, raising if init() has not run"""
|
"""return the default instance, raising if init() has not run"""
|
||||||
if _default is None:
|
if _default is None:
|
||||||
raise RuntimeError("mongo not initialized; call mongo.init(conn, db) first")
|
raise RuntimeError("mongo not initialized; call mongo.init(conn, db) first")
|
||||||
|
|||||||
Reference in New Issue
Block a user