fastapi_contrib.db package

Submodules

fastapi_contrib.db.client module

class fastapi_contrib.db.client.MongoDBClient

Bases: object

Singleton client for interacting with MongoDB. Operates mostly using models, specified when making DB queries.

Implements only part of internal motor methods, but can be populated more

Please don’t use it directly, use fastapi_contrib.db.utils.get_db_client.

async count(model: fastapi_contrib.db.models.MongoDBModel, session: Optional[pymongo.client_session.ClientSession] = None, **kwargs) → int

async delete(model: fastapi_contrib.db.models.MongoDBModel, session: Optional[pymongo.client_session.ClientSession] = None, **kwargs) → pymongo.results.DeleteResult

async get(model: fastapi_contrib.db.models.MongoDBModel, session: Optional[pymongo.client_session.ClientSession] = None, **kwargs) → dict

get_collection(collection_name: str) → pymongo.collection.Collection

async insert(model: fastapi_contrib.db.models.MongoDBModel, session: Optional[pymongo.client_session.ClientSession] = None, include=None, exclude=None) → pymongo.results.InsertOneResult

list(model: fastapi_contrib.db.models.MongoDBModel, session: Optional[pymongo.client_session.ClientSession] = None, _offset: int = 0, _limit: int = 0, _sort: Optional[list] = None, **kwargs) → pymongo.cursor.Cursor

async update_many(model: fastapi_contrib.db.models.MongoDBModel, filter_kwargs: dict, session: Optional[pymongo.client_session.ClientSession] = None, **kwargs) → pymongo.results.UpdateResult

async update_one(model: fastapi_contrib.db.models.MongoDBModel, filter_kwargs: dict, session: Optional[pymongo.client_session.ClientSession] = None, **kwargs) → pymongo.results.UpdateResult

fastapi_contrib.db.models module

class fastapi_contrib.db.models.MongoDBModel(*, id: int = None)

Bases: pydantic.main.BaseModel

Base Model to use for any information saving in MongoDB. Provides id field as a base, populated by id-generator. Use it as follows:

class MyModel(MongoDBModel):
    additional_field1: str
    optional_field2: int = 42

    class Meta:
        collection = "mymodel_collection"


mymodel = MyModel(additional_field1="value")
mymodel.save()

assert mymodel.additional_field1 == "value"
assert mymodel.optional_field2 == 42
assert isinstance(mymodel.id, int)

class Config

Bases: object

anystr_strip_whitespace = True

async classmethod count(**kwargs) → int

async classmethod create_indexes() → Optional[List[str]]

async classmethod delete(**kwargs) → pymongo.results.DeleteResult

async classmethod get(**kwargs) → Optional[fastapi_contrib.db.models.MongoDBModel]

classmethod get_db_collection() → str

id: int

async classmethod list(raw=True, _limit=0, _offset=0, _sort=None, **kwargs)

async save(include: set = None, exclude: set = None, rewrite_fields: dict = None) → int

classmethod set_id(v, values, **kwargs) → int

If id is supplied (ex. from DB) then use it, otherwise generate new.

async classmethod update_many(filter_kwargs: dict, **kwargs) → pymongo.results.UpdateResult

async classmethod update_one(filter_kwargs: dict, **kwargs) → pymongo.results.UpdateResult

class fastapi_contrib.db.models.MongoDBTimeStampedModel(*, id: int = None, created: datetime.datetime = None)

Bases: fastapi_contrib.db.models.MongoDBModel

TimeStampedModel to use when you need to have created field, populated at your model creation time.

Use it as follows:

class MyTimeStampedModel(MongoDBTimeStampedModel):

    class Meta:
        collection = "timestamped_collection"


mymodel = MyTimeStampedModel()
mymodel.save()

assert isinstance(mymodel.id, int)
assert isinstance(mymodel.created, datetime)

created: datetime.datetime

classmethod set_created_now(v: datetime.datetime) → datetime.datetime

If created is supplied (ex. from DB) -> use it, otherwise generate new.

class fastapi_contrib.db.models.NotSet

Bases: object

fastapi_contrib.db.serializers module

fastapi_contrib.db.utils module

async fastapi_contrib.db.utils.create_indexes() → List[str]

Gets all models in project and then creates indexes for each one of them. :return: list of indexes that has been invoked to create

(could’ve been created earlier, it doesn’t raise in this case)

fastapi_contrib.db.utils.default_id_generator(bit_size: int = 32) → int

Generator of IDs for newly created MongoDB rows.

Returns

bit_size long int

fastapi_contrib.db.utils.get_db_client()

Gets instance of MongoDB client for you to make DB queries. :return: MongoDBClient

fastapi_contrib.db.utils.get_models() → list

Scans settings.apps_folder_name. Find models modules in each of them and get all attributes there. Last step is to filter attributes to return only those, subclassed from MongoDBModel (or timestamped version).

Used internally only by create_indexes function.

Returns

list of user-defined models (subclassed from MongoDBModel) in apps

fastapi_contrib.db.utils.get_next_id() → int

Retrieves ID generator function from the path, specified in project’s conf. :return: newly generated ID

fastapi_contrib.db.utils.setup_mongodb(app: fastapi.applications.FastAPI) → None

Helper function to setup MongoDB connection & motor client during setup. Use during app startup as follows:

app = FastAPI()

@app.on_event('startup')
async def startup():
    setup_mongodb(app)

Parameters

app – app object, instance of FastAPI

Returns

None

Module contents