init commit

.gitignore

+.idea/
+__pycache__/
+**.pyc
+

README.md

+Cake in a bot bot

app.py

+""" Bot App """
+import json
+import sys
+import traceback
+from datetime import datetime
+from http import HTTPStatus
+
+import marshmallow_dataclass
+from aiohttp import web
+from aiohttp.web import Request, Response, json_response
+from azure.cosmos import PartitionKey
+from botbuilder.core import (
+    BotFrameworkAdapterSettings,
+    TurnContext,
+    BotFrameworkAdapter,
+)
+from botbuilder.schema import Activity, ActivityTypes
+from marshmallow import ValidationError, EXCLUDE
+
+from bots import TeamsMessagingExtensionsActionPreviewBot
+from bots.exceptions import ConversationNotFound, DataParsingError
+from config import AppConfig, COSMOS_CLIENT, CosmosDBConfig
+from entities.json.notification import Notification, NotificationCosmos
+from utils.cosmos_client import ItemNotFound
+from utils.json_func import json_loads
+
+app_config = AppConfig()
+
+# Create adapter.
+# See https://aka.ms/about-bot-adapter to learn more about how bots work.
+app_settings = BotFrameworkAdapterSettings(app_config.APP_ID,
+                                           app_config.APP_PASSWORD)
+
+ADAPTER = BotFrameworkAdapter(app_settings)
+
+
+# noinspection PyShadowingNames
+async def on_error(context: TurnContext, error: Exception):
+    """ Executed on any error """
+    # This check writes out errors to console log .vs. app insights.
+    # NOTE: In production environment,
+    #       you should consider logging this to Azure application insights.
+    print(f"\n [on_turn_error] unhandled error: {error}", file=sys.stderr)
+    traceback.print_exc()
+
+    # Send a message to the user
+    await context.send_activity(
+        "The bot encountered an error or bug.\r\n"
+        "To continue to run this bot, please fix the bot source code."
+    )
+
+    # Send a trace activity if we're talking to the Bot Framework Emulator
+    if context.activity.channel_id == "emulator":
+        # Create a trace activity that contains the error object
+        trace_activity = Activity(
+            label="TurnError",
+            name="on_turn_error Trace",
+            timestamp=datetime.utcnow(),
+            type=ActivityTypes.trace,
+            value=f"{error}",
+            value_type="https://www.botframework.com/schemas/error",
+        )
+        # Send a trace activity,
+        # which will be displayed in Bot Framework Emulator
+        await context.send_activity(trace_activity)
+
+
+ADAPTER.on_turn_error = on_error
+BOT = TeamsMessagingExtensionsActionPreviewBot(app_settings, ADAPTER)
+
+
+async def v1_get_initiations(request: Request) -> Response:
+    """ Get Initiations by Notification ID """
+    # noinspection PyBroadException
+    try:
+        notification_id = request.match_info['notification_id']
+        inits = await COSMOS_CLIENT.get_initiation_items(notification_id)
+        data = dict(data=[dict(initiator=init.initiator,
+                               timestamp=init.timestamp,
+                               id=init.id) for init in inits])
+        return Response(body=json.dumps(data), status=HTTPStatus.OK)
+    except ItemNotFound as e:
+        print("ItemNotFound:", e)
+        return Response(status=HTTPStatus.NOT_FOUND)
+    except Exception as e:
+        print("error:", e)
+    return Response(status=HTTPStatus.BAD_REQUEST)
+
+
+async def v1_get_notification(request: Request) -> Response:
+    """ Get Notification by ID """
+    # noinspection PyBroadException
+    try:
+        notification_id = request.match_info['notification_id']
+        notification = await COSMOS_CLIENT.get_notification(notification_id)
+        acks = await COSMOS_CLIENT.get_acknowledge_items(notification_id)
+        data = dict(data=dict(
+            timestamp=notification.timestamp,
+            status="DELIVERED",
+            acknowledged=[dict(username=ack.username,
+                               timestamp=ack.timestamp) for ack in acks],
+        ))
+        return Response(body=json.dumps(data), status=HTTPStatus.OK)
+    except ItemNotFound as e:
+        print("ItemNotFound:", e)
+        return Response(status=HTTPStatus.NOT_FOUND)
+    except Exception as e:
+        print("error:", e)
+    return Response(status=HTTPStatus.BAD_REQUEST)
+
+
+async def v1_notification(request: Request) -> Response:
+    """ Notify channel with the link """
+    # todo(s1z): add auth
+    # noinspection PyBroadException
+    try:
+        request_body = await request.text()
+        schema = Notification.get_schema(unknown=EXCLUDE)
+        notification = schema.load(json_loads(request_body, {})).to_db()
+        message_id = await BOT.send_notification(notification)
+        response_body = json.dumps({"data": {"messageId": message_id}})
+        return Response(body=response_body, status=HTTPStatus.OK)
+    except ConversationNotFound:
+        return Response(status=HTTPStatus.NOT_FOUND,
+                        reason="Conversation not found")
+    except DataParsingError:
+        return Response(status=HTTPStatus.BAD_REQUEST,
+                        reason="Bad data structure")
+    except Exception as e:
+        print("error:", e)
+        return Response(status=HTTPStatus.INTERNAL_SERVER_ERROR)
+
+
+async def v1_messages(request: Request) -> Response:
+    """ messages endpoint """
+    if "application/json" in request.headers["Content-Type"]:
+        body = await request.json()
+    else:
+        return Response(status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
+
+    activity = Activity().deserialize(body)
+    auth_header = (request.headers["Authorization"]
+                   if "Authorization" in request.headers else "")
+
+    invoke_response = await ADAPTER.process_activity(
+        activity, auth_header, BOT.on_turn
+    )
+    if invoke_response:
+        return json_response(data=invoke_response.body,
+                             status=invoke_response.status)
+    return Response(status=HTTPStatus.OK)
+
+
+@web.middleware
+async def error_middleware(request, handler):
+    """ Error handler """
+    try:
+        response = await handler(request)
+        if response.status != 404:
+            return response
+        message = response.reason
+    except web.HTTPException as ex:
+        if ex.status != 404:
+            raise
+        message = ex.reason
+    return Response(status=HTTPStatus.NOT_FOUND,
+                    body=json.dumps({"error": message}))
+
+
+APP = web.Application(middlewares=[error_middleware])
+APP.router.add_post("/api/v1/messages", v1_messages)
+APP.router.add_post("/api/v1/notification", v1_notification)
+APP.router.add_get("/api/v1/notification/{notification_id}",
+                   v1_get_notification)
+APP.router.add_get("/api/v1/initiations/{notification_id}", v1_get_initiations)
+
+
+BOT.add_web_app(APP)
+BOT.add_cosmos_client(COSMOS_CLIENT)
+
+
+if __name__ == "__main__":
+    try:
+        web.run_app(APP, host="0.0.0.0", port=app_config.PORT)
+    except Exception as error:
+        raise error

assets/cards/default_card.json

+{
+    "type": "AdaptiveCard",
+    "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+    "version": "1.5",
+    "body": [
+        {
+            "type": "Container",
+            "items": [
+                {
+                    "type": "TextBlock",
+                    "wrap": true,
+                    "text": "Open task module card"
+                }
+            ]
+        },
+        {
+            "type": "Container",
+            "items": [
+                {
+                    "type": "ActionSet",
+                    "actions": [
+                        {
+                            "type": "Action.Submit",
+                            "title": "Open Task",
+                            "data": {
+                                "msteams": {"type": "task/fetch"},
+                                "mx": {
+                                    "type": "task/default"
+                                }
+                            }
+                        }
+                    ]
+                }
+            ]
+        }
+    ]
+}
\ No newline at end of file

bots/__init__.py

+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from .messaging_extension_action_preview_bot import (
+    TeamsMessagingExtensionsActionPreviewBot,
+)
+
+__all__ = ["TeamsMessagingExtensionsActionPreviewBot"]

bots/exceptions.py

+""" Bot Exceptions """
+
+
+class BotException(Exception):
+    """ Base Bot Exception """
+    def __init__(self, message="BotError"):
+        self.message = message
+
+
+class ConversationNotFound(BotException):
+    """ Conversation Not found exception """
+    pass
+
+
+class DataParsingError(BotException):
+    """ Data Parsing Error """
+    pass

config.py

+""" Config """
+import os
+
+from azure.cosmos import PartitionKey
+
+from utils.cosmos_client import CosmosClient
+
+
+PROJECT_ROOT_PATH = os.path.dirname(os.path.abspath("__file__"))
+CARDS_PATH = os.path.join(PROJECT_ROOT_PATH, "assets/cards")
+
+
+class TaskModuleConfig:
+    """ Task Module config """
+    TITLE = os.environ.get("TASK_MODULE_TITLE",
+                           "Example portal")
+    URL = os.environ.get("TASK_MODULE_URL",
+                         "https://fake.s1z.info/show-channel.html")
+    WIDTH = "large"
+    HEIGHT = "large"
+
+
+class AppConfig:
+    """ Bot Configuration """
+
+    PORT = 3978
+    APP_ID = os.environ.get("MS_APP_ID",
+                            "d472f12a-323b-4058-b89b-7a4b15c48ab7")
+    APP_PASSWORD = os.environ.get("MS_APP_PASSWORD",
+                                  "ZdL|zw:]Io_}goFfWs2{w70g+.FXwQ")
+    TENANT_ID = os.environ.get("TENANT_ID",
+                               "5df91ebc-64fa-4aa1-862c-bdc0cba3c656")
+
+
+class CosmosDBConfig:
+    """ Cosmos Databases """
+    HOST = os.environ.get('ACCOUNT_HOST',
+                          'https://nancycosomsdb.documents.azure.com:443/')
+    KEY = os.environ.get('ACCOUNT_KEY',
+                         'fNVRCesO1NAb9MYZNK2rKdAPkY9J4O5ntR8CRuKu6wVGhndiaXch'
+                         'Q6fKwrTTnTbv4tPM8S74YjZsfcX4uAHgiw==')
+
+    class Conversations:
+        """ Conversation DB """
+        DATABASE = "bot"
+        CONTAINER = "conversations"
+        PK = "id"
+        PARTITION_KEY = PartitionKey(path="/conversation/tenantId")
+
+    class Notifications:
+        """ Notifications DB """
+        DATABASE = "bot"
+        CONTAINER = "notifications"
+        PK = "id"
+        PARTITION_KEY = PartitionKey(path="/tenantId")
+
+    class Acknowledges:
+        """ Acknowledges"""
+        DATABASE = "bot"
+        CONTAINER = "acknowledges"
+        PK = "id"
+        PARTITION_KEY = PartitionKey(path="/notificationId")
+
+    class Initiations:
+        """ Initiations """
+        DATABASE = "bot"
+        CONTAINER = "initiations"
+        PK = "id"
+        PARTITION_KEY = PartitionKey(path="/notificationId")
+
+
+COSMOS_CLIENT = CosmosClient(CosmosDBConfig.HOST, CosmosDBConfig.KEY)

entities/json/acknowledge.py

+""" Acknowledge object """
+from dataclasses import dataclass, field
+from typing import Optional
+
+from entities.json.camel_case_mixin import CamelCaseMixin, uuid_factory
+
+
+@dataclass
+class Acknowledge(CamelCaseMixin):
+    """ Acknowledge """
+    id: str = field(default_factory=uuid_factory)
+    notification_id: Optional[str] = field(default=None)
+    username: Optional[str] = field(default=None)
+    user_aad_id: Optional[str] = field(default=None)
+    timestamp: Optional[int] = field(default=None)

entities/json/acknowledge_schema.py

+""" Notification Schema Implementation """
+from marshmallow import fields
+
+from .camel_case_schema import CamelCaseSchema
+
+
+class AcknowledgeSchema(CamelCaseSchema):
+    """ Notification Schema """
+    id = fields.String(required=True, allow_none=True)  # database message id
+    notification_id = fields.String(required=True)
+    username = fields.String(required=True)
+    user_aad_id = fields.String(required=True)
+    timestamp = fields.Integer(required=True)

entities/json/camel_case_mixin.py

+""" marshmallow-dataclass need a mixin to work with camel - snake cases """
+import uuid
+from datetime import datetime
+
+import marshmallow_dataclass
+from marshmallow import pre_load, post_dump
+from stringcase import snakecase, camelcase
+
+
+def uuid_factory() -> str:
+    """ Set unique UUID """
+    return uuid.uuid4().__str__()
+
+
+def timestamp_factory() -> int:
+    """ Set current unix timestamp """
+    return int(datetime.utcnow().timestamp() * 1000)
+
+
+class CamelCaseMixin:
+    """ Camel Case mixin """
+
+    @pre_load
+    def to_snake_case(self, data, **_kwargs):
+        """ to snake case pre load method """
+        return {snakecase(key): value for key, value in data.items()}
+
+    @post_dump
+    def to_camel_case(self, data, **_kwargs):
+        """ to camel case post load method """
+        return {camelcase(key): value for key, value in data.items()}
+
+    @classmethod
+    def get_schema(cls, *args, **kwargs):
+        """ Get schema """
+        return marshmallow_dataclass.class_schema(cls)(*args, **kwargs)

entities/json/camel_case_schema.py

+""" Camel Case schema implementation """
+from marshmallow import Schema, fields
+
+
+def camelcase(s):
+    """ Convert camel case to snake case"""
+    parts = iter(s.split("_"))
+    return next(parts) + "".join(i.title() for i in parts)
+
+
+class CamelCaseSchema(Schema):
+    """Schema that uses camel-case for its external representation
+    and snake-case for its internal representation.
+    """
+
+    def on_bind_field(self, field_name, field_obj):
+        """ On bind field callback """
+        field_obj.data_key = camelcase(field_obj.data_key or field_name)

entities/json/conversation_reference_schema.py

+""" Conversation Reference Implementation """
+from typing import Dict
+
+from botbuilder.schema import ConversationReference, ConversationAccount, \
+    ChannelAccount
+from marshmallow import fields, EXCLUDE, validate, post_load
+
+from .camel_case_schema import CamelCaseSchema
+
+
+class UserSchema(CamelCaseSchema):
+    """ User Schema """
+    id = fields.String(required=True)
+    name = fields.String(required=True)
+    # may be null if it's a bot
+    aad_object_id = fields.String(required=True, allow_none=True)
+    role = fields.String(required=True, allow_none=True)
+
+
+class ConversationSchema(CamelCaseSchema):
+    """ Conversation Schema """
+    is_group = fields.Boolean(required=True, allow_none=True)
+    conversation_type = fields.String(required=True)
+    id = fields.String(required=True)
+    name = fields.String(required=True, allow_none=True)
+    aad_object_id = fields.String(required=True, allow_none=True)
+    role = fields.String(required=True, allow_none=True)
+    tenant_id = fields.String(required=True)
+    # Microsoft does not describe this object at all
+    properties = fields.Dict(required=False, allow_none=True)
+
+
+class ConversationReferenceSchema(CamelCaseSchema):
+    """ Conversation Reference schema """
+
+    activity_id = fields.String()
+    user = fields.Nested(UserSchema, required=False, unknown=EXCLUDE)
+    bot = fields.Nested(UserSchema, unknown=EXCLUDE)
+    conversation = fields.Nested(ConversationSchema, unknown=EXCLUDE)
+    channel_id = fields.String(required=True)
+    locale = fields.String(required=True)
+    service_url = fields.String()
+
+    @post_load
+    def create_conversation_reference(self, data, **_kwargs):
+        """ Create Conversation Reference """
+        data.update(dict(
+            user=ChannelAccount(**data.pop("user")),
+            bot=ChannelAccount(**data.pop("bot")),
+            conversation=ConversationAccount(**data.pop("conversation"))
+        ))
+        return ConversationReference(**data)

entities/json/initiation.py

+""" Initiation object """
+from dataclasses import dataclass, field
+from typing import Optional
+
+
+from entities.json.camel_case_mixin import CamelCaseMixin
+
+
+@dataclass
+class Initiation(CamelCaseMixin):
+    """ Notification Dataclass """
+    initiator: str  # User name
+    notification_id: str  # Notification ID
+    timestamp: Optional[int] = field(default=None)
+    id: Optional[str] = field(default=None)  # Unique Initiation ID

entities/json/medx.py

+""" Notification object """
+from dataclasses import dataclass, field
+from typing import Optional
+
+from entities.json.camel_case_mixin import CamelCaseMixin
+
+
+class MXTypes:
+    """ MedX Types """
+    UNKNOWN = "UNKNOWN"
+    ACKNOWLEDGE = "acknowledge"
+
+    class Task:
+        """ Task types """
+        DEFAULT = "task/default"
+        NOTIFICATION = "task/notification"
+
+
+@dataclass
+class MedX(CamelCaseMixin):
+    """ MedX data """
+    type: str
+    notification_id: Optional[str]

entities/json/notification.py

+""" Notification object """
+from dataclasses import dataclass, field
+from typing import Optional
+
+import marshmallow.validate
+
+from entities.json.camel_case_mixin import CamelCaseMixin, timestamp_factory
+
+
+@dataclass
+class NotificationUrl(CamelCaseMixin):
+    """ Notifiction URL """
+    title: Optional[str]
+    link: Optional[str] = field(metadata=dict(
+        validate=marshmallow.validate.URL()
+    ))
+
+
+@dataclass
+class Notification(CamelCaseMixin):
+    """ Notification Dataclass """
+    message_id: Optional[str]
+    destination: str
+    subject: Optional[str] = field(default=None)
+    message: Optional[str] = field(default=None)
+    title: Optional[str] = field(default=None)
+    url: Optional[NotificationUrl] = field(default_factory=NotificationUrl)
+    acknowledge: Optional[bool] = field(default=False)
+
+    def to_db(self) -> "NotificationCosmos":
+        """ Create NotificationCosmos """
+        return NotificationCosmos(message_id=self.message_id,
+                                  destination=self.destination,
+                                  subject=self.subject,
+                                  message=self.message,
+                                  title=self.title,
+                                  url=self.url,
+                                  acknowledge=self.acknowledge)
+
+
+# noinspection PyDataclass
+@dataclass
+class NotificationCosmos(Notification):
+    """ Notification Dataclass """
+    # We have to add these fields
+    id: Optional[str] = field(default=None)
+    tenant_id: Optional[str] = field(default=None)
+    timestamp: Optional[int] = field(default_factory=timestamp_factory)

entities/json/notification_schema.py

+""" Notification Schema Implementation """
+from marshmallow import fields, EXCLUDE
+
+from .camel_case_schema import CamelCaseSchema
+
+
+class NotificationURLSchema(CamelCaseSchema):
+    """ Notification URL Schema """
+    title = fields.String()
+    link = fields.String()
+
+
+# TODO(s1z): Migrate to marshmallow-dataclass
+class NotificationSchema(CamelCaseSchema):
+    """ Notification Schema """
+    id = fields.String(required=True, allow_none=True)  # database message id
+    tenant_id = fields.String(required=True, allow_none=True)
+    destination = fields.String(required=True)
+    message_id = fields.String()  # teams message id
+    subject = fields.String()
+    message = fields.String()
+    title = fields.String()
+    url = fields.Nested(NotificationURLSchema, unknown=EXCLUDE)
+    acknowledge = fields.Boolean(default=False)

requirements.txt

+botbuilder-integration-aiohttp>=4.14.0
+azure-cosmos==4.2.0
+marshmallow==3.15.0
+aiohttp~=3.8.1
+simplejson==3.17.6
+marshmallow-dataclass==8.5.8
+stringcase==1.2.0

utils/card_helper.py

+import os
+from typing import Dict, Any, Optional, Mapping
+
+from config import CARDS_PATH
+from entities.json.notification import NotificationCosmos
+from utils.json_func import json_loads
+
+
+FILE = "__file__"
+
+
+class CardHelper:
+    """ Card Helper """
+
+    @staticmethod
+    def load_assets_card(name: str) -> Mapping[str, Any]:
+        """ 123 """
+        filename = name + ".json" if name.find(".json") < 0 else name
+        filename_path = os.path.join(CARDS_PATH, filename)
+        with open(filename_path, "r") as f:
+            card_data = f.read()
+            return json_loads(card_data)
+
+    @staticmethod
+    def create_notification_card(notification: NotificationCosmos,
+                                 acknowledged_by: Optional[str] = None)\
+            -> Dict[str, Any]:
+        """ Create notification card """
+
+        title = notification.title or "You got a new notification!"
+        notification_id = notification.id or None
+        subject = notification.subject or None
+        message = notification.message or None
+        url = notification.url or None
+        acknowledge = notification.acknowledge
+        card_body = []
+
+        # ======================= TITLE =======================================
+        if title is not None:
+            card_body.append({"type": "TextBlock",
+                              "size": "Medium",
+                              "weight": "Bolder",
+                              "text": title})
+
+        # ======================= SUBJ and MESSAGE ============================
+        if subject is not None:
+            card_body.append({"type": "FactSet",
+                              "facts": [{"title": "Subject:",
+                                         "value": subject}]})
+
+        # ======================= MESSAGE =================================== #
+        if message is not None:
+            card_body.append({"type": "FactSet",
+                              "facts": [{"title": "Message:",
+                                         "value": message}]})
+
+        # ======================= URL =========================================
+        if url is not None and notification_id is not None:
+            url_title = url.title or "Open Notification"
+            card_body.append({
+                "type": "ActionSet",
+                "actions": [{"type": "Action.Submit",
+                             "title": url_title,
+                             "data": {
+                                 "msteams": {
+                                     "type": "task/fetch"
+                                 },
+                                 "mx": {
+                                     "type": "task/notification",
+                                     "notificationId": notification_id
+                                 }
+                             }}]
+            })
+
+        # ======================= URL =========================================
+        if acknowledge is not None and acknowledge and acknowledged_by is None:
+            card_body.append({
+                "type": "ActionSet",
+                "actions": [{"type": "Action.Submit",
+                             "title": "Acknowledge",
+                             "data": {
+                                 "mx": {
+                                     "type": "acknowledge",
+                                     "notificationId": notification_id
+                                 }
+                             }}]
+            })
+        if acknowledged_by is not None:
+            card_body.append({"type": "FactSet",
+                              "facts": [{"title": "Acknowledged:",
+                                         "value": acknowledged_by}]})
+
+        return {
+            "type": "AdaptiveCard",
+            "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+            "version": "1.5",
+            "body": card_body
+        }

utils/function.py

+""" Handy Functions """
+from typing import List, Optional, Dict
+
+
+def get_first_or_none(items: List) -> Optional[Dict[str, any]]:
+    """ Get first object from list or return None len < 1 """
+    if len(items) > 0:
+        return items[0]
+    return None

utils/json_func.py

+""" JSON helper module """
+import sys
+from typing import Any, Union, Optional, Mapping, Iterable
+
+import simplejson as json
+import simplejson.scanner as json_scanner
+
+from utils.log import Log
+
+
+TAG = __name__
+
+
+def json_loads(data: Union[str, bytes], default: Optional[Any] = None) -> \
+        Union[Mapping[str, Any], Iterable[Mapping[str, Any]]]:
+    """ Json.loads wrapper, tries to load data and prints errors if any """
+    try:
+        j_data = json.loads(data, strict=False)
+        if isinstance(j_data, dict) or isinstance(j_data, list):
+            return j_data
+    except TypeError:
+        Log.e("json_loads", "TypeError:", sys.exc_info())
+    except json_scanner.JSONDecodeError:
+        Log.e("json_loads", "JSONDecodeError:", sys.exc_info())
+    return default
+
+
+def json_dumps(*args: Any, **kwargs: Mapping[str, Any]) -> str:
+    """ Json.dumps wrapper, tries to dump data and prints errors if any """
+    try:
+        return json.dumps(*args, **kwargs)
+    except Exception:
+        Log.e("json_loads", "error:", exc_info=sys.exc_info())
+        raise

utils/log.py

+""" Log helper module """
+import logging
+import traceback
+
+
+class Log:
+    """ Logger. This is a pretty useful 'Java style' logger """
+
+    @staticmethod
+    def log(level, source, message="", exc_info=None):
+        """ log method """
+        logger = logging.getLogger()
+        line = "{source}{message}{exc}"
+        exc = ''
+        if isinstance(exc_info, (list, tuple)):
+            ex_type, ex_value, ex_traceback = exc_info
+            exc = ": " + ''.join(
+                traceback.format_exception(ex_type, ex_value, ex_traceback)
+            )
+        message = "::{}".format(message) if message else ""
+        logger.log(level, line.format(source=source, message=message, exc=exc))
+
+    @staticmethod
+    def w(source, message="", exc_info=None):
+        """ warning level """
+        return Log.log(logging.WARN, source, message, exc_info)
+
+    @staticmethod
+    def d(source, message="", exc_info=None):
+        """ debug level """
+        return Log.log(logging.DEBUG, source, message, exc_info)
+
+    @staticmethod
+    def i(source, message="", exc_info=None):
+        """ info level """
+        return Log.log(logging.INFO, source, message, exc_info)
+
+    @staticmethod
+    def e(source, message="error", exc_info=None):
+        """ error level """
+        return Log.log(logging.ERROR, source, message, exc_info)