async_mongo.py

#!/usr/bin/env python3
'''
@desc    Simple MongoDB Client class. Offers the entire API
         of AsyncioMotorCollection.
@author  SDQ <[email protected]>
@version 1.0.0
@date    2020-02-17
@note    1.0.0 (2020-02-17) : First functional version
'''
from typing import Any, Optional
import asyncio
from motor.motor_asyncio import (
    AsyncIOMotorClient,
    AsyncIOMotorCollection,
    AsyncIOMotorCursor,
    AsyncIOMotorDatabase,
)


class MongoDatabase:
    '''Simple MongoDB Client class. Offers the entire API
    of AsyncioMotorCollection.
    
    The purpose of this class is to enable a quick interface to a MongoDB
    database without all the fuss of writing the entire class.

    Please refer to
    https://motor.readthedocs.io/en/stable/api-asyncio/asyncio_motor_collection.html
    for a detailled documentation.

    This class has NOT been massively tested.
    '''  # noqa

    __version__: str = '1.0.0'

    def __init__(
        self,
        host: str,
        user: str,
        password: str,
        *,
        database: str,
        collection: str = 'data',
        port: int = 27017,
        auth_source: str = 'admin',
        io_loop: Optional[asyncio.AbstractEventLoop] = None,
    ) -> None:
        '''Constructor'''
        self._client: AsyncIOMotorClient = AsyncIOMotorClient(
            f'mongodb://{user}:{password}@{host}:{port}/'
            f'?authSource={auth_source}',
            io_loop=io_loop or asyncio.get_event_loop(),
        )
        self._database: AsyncIOMotorDatabase = self._client[database]
        self._collection: AsyncIOMotorCollection = self._database[collection]

    def __getattr__(self, attr: str) -> Any:
        '''Return the attribute of the AsyncIOMotorCollection if exists.'''
        if attr in dir(self._collection):
            return getattr(self._collection, attr)
        else:
            raise AttributeError(
                f"'{self.__class__.__name__}' object has no attribute '{attr}'"
            )