Module utils

Utils for mailcow

build_attributes(**kwargs)

Convert argparse arguments into a dict for edit and add requests.

Examples:

moo = MailCow() sections = MailCowOpenApi(f'{moo.url}/api/openapi.yaml') args = menu(sections.endpoints) build_attributes(**(vars(args)), routes=sections) will return output dependend on argparse build_attributes( section=alias, modifier=edit, items=["5"], active=True, address=example@example.com, goto=goto@example.com routes=sections) will return: { 'active': "1", 'address': 'example@example.com', 'goto': 'goto@example.com' }

Source code in mailcow/utils.py

def build_attributes(**kwargs):
    '''
    Convert argparse arguments into a dict for edit and add requests.

    Arguments:
        **(vars(args))      This function takes all arguments from argparse
        endpoints    (dict) Object provided by MailCowOpenApi()

    Example:
        moo = MailCow()
        sections = MailCowOpenApi(f'{moo.url}/api/openapi.yaml')
        args = menu(sections.endpoints)
        build_attributes(**(vars(args)), routes=sections)
        will return output dependend on argparse
        build_attributes(
            section=alias,
            modifier=edit,
            items=["5"],
            active=True,
            address=example@example.com,
            goto=goto@example.com
            routes=sections)
        will return: {
            'active': "1",
            'address': 'example@example.com',
            'goto': 'goto@example.com'
        }
    '''
    endpoints = kwargs.get('endpoints')
    section = kwargs.get('section')
    modifier = kwargs.get('modifier')
    attrs = dict()

    for arg in endpoints[section][modifier]:
        value = kwargs.get(arg)

        # Bools must be converted into int str
        if isinstance(value, bool):
            # Hack avoid using argparse.BooleanOptionalAction as action
            if f'--no-{arg}' in sys.argv:
                value = False
            attrs.update({arg: str(int(value))})
        else:
            attrs.update({arg: value})
        # items are usually stored in separate list
        if value is None or arg == 'items':
            del attrs[arg]

    return attrs

chomp(data)

Remove multi-whitespaces and newlines

Source code in mailcow/utils.py

def chomp(data):
    '''Remove multi-whitespaces and newlines'''
    return re.sub(r'[ ]{2,}', '', data.strip())

debug_mailcow(debug=False)

Set loglevel to debug and log to stdout

Source code in mailcow/utils.py

def debug_mailcow(debug=False):
    '''Set loglevel to debug and log to stdout'''
    logger = logging.getLogger()
    if debug:
        logger.setLevel(logging.DEBUG)
    logging.StreamHandler(sys.stdout)

debug_msg(data)

Debug what's happening

Source code in mailcow/utils.py

def debug_msg(data):
    '''Debug what's happening'''
    if isinstance(data, str):
        logging.debug(chomp(data))
    else:
        logging.debug(data)

describeOpenApiPath(path)

Parse path from openapi and return results as dict

Source code in mailcow/utils.py

def describeOpenApiPath(path):
    '''Parse path from openapi and return results as dict'''
    path = list(filter(None, path.split('/')[3:]))
    modifier = path[0]
    section = path[1]
    parm_all = False
    component = 'no_log' if path[-1] == 'no_log' else None
    parameter = path[-1] if '{' in path[-1] else None

    if len(path) > 2:
        parm_all = (path[2] == 'all')
        if '{' not in path[2] and path[2] != 'all':
            component = path[2]

    return dict(
        modifier=modifier,
        section=section,
        all=parm_all,
        component=component,
        parameter=parameter)

filterOpenApiPath(schema)

filter out relevant dicts from OpenApi

Source code in mailcow/utils.py

def filterOpenApiPath(schema):
    '''filter out relevant dicts from OpenApi'''
    method = ''.join(schema)
    parameters = schema[method].get('parameters')
    requestbody = schema[method].get('requestBody')
    schema = {}
    if requestbody:
        schema = requestbody['content']['application/json']['schema']

    return dict(
        parameters=parameters,
        schema=schema
    )

getOpenApiParameters(data)

Extract data from OpenApi that is passed via URL.

Source code in mailcow/utils.py

def getOpenApiParameters(data):
    '''Extract data from OpenApi that is passed via URL.'''
    arguments = dict()

    if isinstance(data, list):
        for parameter in data:
            if parameter['in'] == 'path':
                arguments.update({
                    parameter['name']: {
                        'description': parameter['description'],
                        'type': parameter['schema']['type']}})

    if isinstance(data, str):
        # This type isn't 'boolean' on purpose. argparse will distinguish
        # between store_true and OptionalBoolean this way
        arguments.update({data: {
            'description': f'get {data} entries',
            'type': 'bool'}})

    return arguments

getOpenApiProperties(schema)

Extract data from OpenApi that must be send via body.

Source code in mailcow/utils.py

def getOpenApiProperties(schema):
    '''Extract data from OpenApi that must be send via body.'''
    properties = schema.get('properties', {})
    items = schema.get('items')
    attr = properties.get('attr')

    # schema/items instead of schema/properties/items
    # missing in delete/{alias,dkim}
    if items:
        properties.update({'items': schema['items']})

    # move properties to top and remove redundant properties key
    if attr:
        properties.update(attr['properties'])
        del properties['attr']

    # https://github.com/mailcow/mailcow-dockerized/blob/master/data/web/api/openapi.yaml#L1024
    # https://github.com/mailcow/mailcow-dockerized/blob/master/data/web/api/openapi.yaml#L3055
    # There is a typo in the mailbox update,add properties
    if 'pasword' in properties.keys():
        properties.update({'password': properties['pasword']})
        del properties['pasword']

    return properties

parse_fields(fields, vertical)

Splitts fields if necessary

Examples:

Incase someone use fields as commaseparated list ie: [...] --fields address,goto instead of [...] --fields address --fields goto

Source code in mailcow/utils.py

def parse_fields(fields, vertical):
    '''
    Splitts `fields` if necessary

    Attributes:
        fields      (string)  Will be converted to list if splittable by ','
        vertical    (boolean) Skip splitting if True

    Example:
        Incase someone use fields as commaseparated list ie:
        `[...] --fields address,goto` instead of
        `[...] --fields address --fields goto`
    '''
    if fields and not vertical:
        for field in fields:
            fields = field.split(',') if ',' in field else fields
        debug_msg(f'Filter Fields: {fields}')

    return fields

prepare_getRequest(**kwargs)

Sections in /api​/v1​/get​/ require more manipulation. This will compare argparse values with openapi parameters and setup url accordingly.

Examples:

moo = MailCow() sections = MailCowOpenApi(f'{moo.url}/api/openapi.yaml')

prepare_getRequest(section='syncjobs', all=True, no_log=True) will return syncjobs​/all​/no_log prepare_getRequest(section='logs', api=True, count=5) will return logs/api/5

Source code in mailcow/utils.py

def prepare_getRequest(**kwargs):
    '''
    Sections in /api​/v1​/get​/ require more manipulation.
    This will compare argparse values with openapi parameters
    and setup url accordingly.

    Arguments:
        **(vars(args))     This function takes all arguments from argparse
        endpoints    (dict) Object provided by MailCowOpenApi()

    Example:
    moo = MailCow()
    sections = MailCowOpenApi(f'{moo.url}/api/openapi.yaml')

    prepare_getRequest(section='syncjobs', all=True, no_log=True)
    will return `syncjobs​/all​/no_log`
    prepare_getRequest(section='logs', api=True, count=5)
    will return `logs/api/5`
    '''
    endpoints = kwargs.get('endpoints')
    section = kwargs.get('section')
    return_section = section
    arguments = endpoints[section]['get']
    count = kwargs.get('count')
    mailbox = kwargs.get('mailbox')

    for argument in arguments.keys():
        if arguments[argument]['type'] == 'bool' and kwargs.get(argument):
            return_section = f'{section}/{argument}'
            if count:
                return_section = f'{return_section}/{count}'

    if kwargs.get('all'):
        return_section = f'{section}/all'

        # its always /api​/v1​/get​/syncjobs​/all​/no_log
        if kwargs.get('no_log'):
            return_section = f'{return_section}/no_log'

    # can be: /api​/v1​/get​/app-passwd/all​/$mailbox
    # or:/api​/v1​/get​/rl-mbox​/$mailbox
    if mailbox:
        return_section = f'{return_section}/{mailbox}'

    # if set id and domain should override return_section
    if kwargs.get('id'):
        return_section = f'{section}/{kwargs["id"]}'
    if kwargs.get('domain'):
        return_section = f'{section}/{kwargs["domain"]}'

    debug_msg(f'Parsed get request: {return_section}')

    return return_section

validate_response(func)

Decorator validating session responses

Source code in mailcow/utils.py

def validate_response(func):
    '''Decorator validating session responses'''
    def validate(*args, **kwargs):
        response = func(*args, **kwargs)
        for msg in [f'Request Status Code: {response.status_code}',
                    f'Request Headers: {response.headers}']:
            debug_msg(msg)

        color = '\x1b[0m'
        if response.status_code < 400:
            color = '\x1b[6;30;42m'  # green
        if response.status_code >= 400:
            color = '\x1b[6;30;43m'  # yellow
        if response.status_code >= 500:
            color = '\x1b[6;30;41m'  # red
        end = '\x1b[0m'

        debug_msg(f'API response {response.url}:'
                  f'{color} {response.status_code} {end}')

        if 'json' in response.headers['Content-Type'] and \
                len(response.content) > 0:
            body = response.json()
        else:
            body = response.content

        debug_msg(f'Request Content: {body}')
        return body

    return validate

---

Last update: March 22, 2021