This skill should be used when writing repository methods, performing MongoDB queries with Motor, handling upserts, preventing duplicate array entries, or designing new query patterns that require indexing.
async def for repository methods — Motor is fully async.await on all Motor operations: find_one, find, update_one, insert_one.config.py.ValueError (not HTTPException) from repos.Partial updates — always use $set to avoid overwriting unrelated fields:
await collection.update_one(
{"video_id": video_id},
{"$set": {"dubbed_url": url, "updated_at": datetime.utcnow()}}
)
— create or update atomically:
await collection.update_one(
{"project_id": project_id},
{"$set": data},
upsert=True
)
Prevent duplicate array entries — filter the outer query with $ne on the unique key so the push only happens when the entry is absent (persist_job_result pattern):
# Only push if job_id not already in array
await collection.update_one(
{"video_id": video_id, "dubbed_versions.job_id": {"$ne": job_id}},
{"$push": {"dubbed_versions": version_data}}
)
When adding a new query pattern, advise on index creation:
# Example: index for frequent lookup by project_id + video_id
await db["videos"].create_index([("project_id", 1), ("video_id", 1)])
Always mention index recommendations when writing new find_one or find queries that filter on non-_id fields.
async def get_video(self, video_id: str) -> dict | None:
doc = await self.collection.find_one({"video_id": video_id})
return doc
# In router:
video = await videos_repo.get_video(video_id)
if not video:
raise HTTPException(status_code=404, detail="Video not found")