[Python] How can I programmatically determine if a PyMongo command is a read or write for...

I'm building an authorization/access control system in a Python application that uses PyMongo to access MongoDB. I need to dynamically decide—at runtime and for arbitrary query shapes—whether a given MongoDB command issued via PyMongo is a read or a write operation, in order to split processing into read-permitted vs. write-permitted users.

Some additional requirements and context:

The app issues arbitrary commands and queries, not just basic .find() or .insert_one() calls. For example, it may run complex aggregations, map-reduce, find_and_modify, etc.

I can't use a hardcoded list of command names to determine operation type because MongoDB adds new commands over time, and commands like aggregate can be read only or write depending on the pipeline.

The use case includes enforcement of permissions at the proxy/middleware level, so it must work before commands are handed off to MongoDB.

I researched the official MongoDB documentation and PyMongo API, but could not find a public, up-to-date method to classify an arbitrary command as "read" or "write". I found that the Node.js driver exposes writable command sets, but PyMongo does not.

After more research (including reading this MongoDB doc: https://www.mongodb.com/docs/manual.../#std-label-replica-set-read-preference-modes and related Stack Overflow answers), I discovered a dynamic approach that leverages MongoDB's replica set behavior:

• Approach: You temporarily send the command to a secondary member of the replica set (using readPreference=SECONDARY). If the command executes without error, it is a read operation. If it fails with "not primary", "NotWritablePrimary", or similar errors/codes, it is a write. The main issue with this would be: what if secondary member doesn't exist when the check is being made, I checked and it would throw an error and so we wouldn't be able to know if its a read or not.
• I implemented a function in Python, using PyMongo and catching NotPrimaryError/OperationFailure, checking error messages and codes. This works in my initial tests, but I want to be sure of correctness and efficiency.
• Specifically, I want to know:
• Is this pattern future-proof and robust for all current and new MongoDB/PyMongo versions?
• Are there any side-effects or safety concerns with running commands (reads) on secondaries this way?
• Are there edge cases where a command could be misclassified, e.g., hybrid commands, user-defined stages, or commands that behave differently on primary/secondary?
• Any advice on integrating this technique for efficient real-time checks (e.g., LRU caching of results per command shape)?

Here's a distilled code sketch of what I'm doing:

from pymongo import MongoClient, ReadPreference
from pymongo.errors import NotPrimaryError, OperationFailure

def is_read_command(db, command):
try:
secondary_db = db.with_options(read_preference=ReadPreference.SECONDARY)
secondary_db.command(command)
return True
except (NotPrimaryError, OperationFailure) as exc:
err_msg = str(exc)
if getattr(exc, "code", None) in (10107, 13435) or "not primary" in err_msg.lower() or "not master" in err_msg.lower():
return False
raise

Continue reading...