Skip to content

Database Connection

In Hyac's endpoint functions, you can easily interact with the database through the context object. The platform provides two ways to operate MongoDB:

  • Asynchronous (Motor): Accessed via context.motor_db, offers higher performance and is the recommended method.
  • Synchronous (PyMongo): Accessed via context.pymongo_db, uses a more traditional syntax, suitable for scenarios where you are not familiar with asynchronous programming.

Using the asynchronous approach allows you to take full advantage of Python's asyncio features, avoiding blocking the entire application while waiting for a database response, thus achieving higher concurrency.

Key Points

  • The function must be defined with async def.
  • Get the asynchronous database instance via context.motor_db.
  • Use the await keyword before all database operations.

Code Example

The following example demonstrates how to perform complete CRUD (Create, Read, Update, Delete) operations using Motor in an endpoint function.

from datetime import datetime
from loguru import logger
from bson import ObjectId

async def handler(context, request, name: str = "World", value: int = 0):
    """
    A complete example of database operations using Motor (asynchronous).
    """

    logger.info(f"[Async] Received parameters: name='{name}', value={value}")
    db = context.motor_db  # Get the asynchronous Motor database client
    demo_collection = db["hyac_demo_async"]

    # 1. CREATE
    doc = {"name": name, "value": value, "createdAt": datetime.utcnow()}
    res = await demo_collection.insert_one(doc)
    inserted_id = res.inserted_id
    logger.info(f"[Async] CREATE: Document inserted, ID: {inserted_id}")

    # 2. READ
    read_doc = await demo_collection.find_one({"_id": inserted_id})
    logger.info(f"[Async] READ: Found document: {read_doc}")

    # 3. UPDATE
    await demo_collection.update_one({"_id": inserted_id}, {"$set": {"status": "updated"}})
    updated_doc = await demo_collection.find_one({"_id": inserted_id})
    logger.info(f"[Async] UPDATE: Document status updated: {updated_doc}")

    # 4. DELETE
    await demo_collection.delete_one({"_id": inserted_id})
    logger.info(f"[Async] DELETE: Document cleaned up")

    return {"status": "ok", "driver": "motor (async)", "inserted_id": str(inserted_id)}

2. Synchronous Operations (PyMongo)

If you are more accustomed to traditional synchronous programming or need to use libraries that do not support asynchronous operations, you can use PyMongo.

Hyac optimizes synchronous operations in the background: although your code is synchronous, FastAPI runs it in a separate thread pool to avoid blocking the main event loop.

Key Points

  • It is still recommended to define the function with async def for consistency.
  • Get the synchronous database instance via context.pymongo_db.
  • Database operations use standard synchronous syntax, with no need for await.

Code Example

```python from datetime import datetime from loguru import logger from bson import ObjectId

async def handler(context, request, name: str = "World", value: int = 0): """ A complete example of database operations using PyMongo (synchronous). """ logger.info(f"[Sync] Received parameters: name='{name}', value={value}") db = context.pymongo_db # Get the synchronous PyMongo database client demo_collection = db["hyac_demo_sync"]

# 1. CREATE
doc = {"name": name, "value": value, "createdAt": datetime.utcnow()}
res = demo_collection.insert_one(doc)
inserted_id = res.inserted_id
logger.info(f"[Sync] CREATE: Document inserted, ID: {inserted_id}")

# 2. READ
read_doc = demo_collection.find_one({"_id": inserted_id})
logger.info(f"[Sync] READ: Found document: {read_doc}")

# 3. UPDATE
demo_collection.update_one({"_id": inserted_id}, {"$set": {"status": "updated"}})
updated_doc = demo_collection.find_one({"_id": inserted_id})
logger.info(f"[Sync] UPDATE: Document status updated: {updated_doc}")

# 4. DELETE
demo_collection.delete_one({"_id": inserted_id})
logger.info(f"[Sync] DELETE: Document cleaned up")

return {"status": "ok", "driver": "pymongo (sync)", "inserted_id": str(inserted_id)}