Core#

class quart_mongo.Mongo(app: Quart | None = None, uri: str | None = None, json_options: JSONOptions | None = None, *args, **kwargs)#

This class is for integrating MongoDB into your Quart application. It intergrates with MongoDB by using Motor and Odmantic. It also provides a few helper functions for working with MongoDB.

Parameters:

  • app – The Quart application to use. Defaults to None.

  • uri – The MongoDB URI to use. This parameter defaults to None and will then use the app config.

  • json_options – Options for JSON encoding.

  • args – Arguments to pass to AIOMotorClient on intialization.

  • kwargs – Extra agrugments to pass to AIOMotorClient on intialization.

cx#

The :class::~quart_mongo.wrappers.AIOMotorClient connected to the MongoDB server.

db#

The :class::~quart_mongo.wrappers.AIOMotorDatabase if the URI used named a database or None otherwise.

odm#

The :class::~quart_mongo.wrappers.AIOEngine if the URI used named a database or None otherwise. This is the class to add, get, or update :class::~odmantic.Models to the database.

init_app(app: Quart, uri: str | None = None, json_options: JSONOptions | None = None, *args, **kwargs) → None#

This function configures Mongo and applies the extension instance with the given application.

It also will get the database name from the URI if it was provided, register the connection function with the app as a before serving function, and register BSON and JSON helpers with the app.

Note that BSON and JSON helpers are only added to the app if they have not been. This is in case of multiple instances of this class.

If you are using quart_schema in your application. Make sure you register quart_schema with the application before this extension. This is so the JSON provider for this extension can be used with the application in lieu of the one with `quart_schema.

Parameters:

  • app – The application to use.

  • uri – The MongoDB URI to use. This parameter defaults to None and will then use the app config.

  • json_options – Options for JSON encoding.

  • args – Arguments to pass to AIOMotorClient on intialization.

  • kwargs – Extra arguments to pass to AIOMotorClient on initialization.

async _connect_db() → None#

Before Serving Function (Private)

This function is registered with application with the init_app and is called by the application before serving any routes.

The purpose of the function is to setup AIOMotorClient and also AIOMotorDatabase and AIOEngine if the MongoDB URI provides the database name.

async send_file_by_name(filename: str, base: str = 'fs', version: int = -1, cache_for: int = 31536000) → Response#

Respond with a file from GridFS using the file name.

Returns an instance of the response_class containing the named file, and implement conditional GET semantics.

@app.route("/uploads/<path:filename>")
async def get_upload(filename):
    return await mongo.send_file_by_name(filename)
Parameters:

  • filename – The filename of the file to return

  • base – The base name of the GridFS collections to use

  • version – If positive, return the Nth revision of the file identified by filename; if negative, return the Nth most recent revision. If no such version exists, return with HTTP status 404.

  • cache_for (int) – Number of seconds that browsers should be instructed to cache responses

async send_file_by_id(id: ObjectId, base: str = 'fs', version: int = -1, cache_for: int = 31536000) → Response#

Respond with a file from GridFS using the file Object id.

Returns an instance of the response_class containing the named file, and implement conditional GET semantics.

@app.route("/uploads/<str:id>")
async def get_upload(id):
    return await mongo.send_file_by_id(id)
Parameters:

  • id – The bson Object id of the file.

  • base – The base name of the GridFS collections to use

  • version – If positive, return the Nth revision of the file identified by filename; if negative, return the Nth most recent revision. If no such version exists, return with HTTP status 404.

  • cache_for (int) – Number of seconds that browsers should be instructed to cache responses

async save_file(filename: str, fileobj: SupportsRead[AnyStr], base: str = 'fs', content_type: str | None = None, **kwargs) → ObjectId#

Save a file-like object to GridFS using the given filename.

@app.route("/uploads/<path:filename>", methods=["POST"])
async def save_upload(filename):
    await mongo.save_file(filename, request.files["file"])
    return redirect(url_for("get_upload", filename=filename))
Parameters:

  • filename – the filename of the file to return

  • fileobj (file) – the file-like object to save

  • base – base the base name of the GridFS collections to use

  • content_type – the MIME content-type of the file. If None, the content-type is guessed from the filename using guess_type()

  • kwargs – extra attributes to be stored in the file’s document,

:param passed directly to gridfs.GridFS.put():