common.helpers.send_mail_to_admins() - Code Metrics - Inspection of ":bug: catch errors when creating a biosample user" - cnr-ibba/IMAGE-InjectTool - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#55)

by Paolo

created 2019-10-01 13:43 UTC

common.helpers.send_mail_to_admins() A

↳ Parent: common.helpers

Complexity

Conditions

Size

Total Lines	21
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	10
dl	0
loc	21
rs	9.9
c	0
b	0
f	0
cc	1
nop	4

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Wed Mar 27 12:50:16 2019

@author: Paolo Cozzi <[email protected]>
"""

import re
import logging
import datetime
import websockets
import time
import json

from dateutil.relativedelta import relativedelta

from django.conf import settings
from django.contrib.admin.utils import NestedObjects
from django.core.mail import send_mass_mail
from django.db import DEFAULT_DB_ALIAS
from django.utils.text import capfirst
from django.utils.encoding import force_text

from .constants import YEARS, MONTHS, DAYS, OBO_URL, TIME_UNITS

# Get an instance of a logger
logger = logging.getLogger(__name__)


def image_timedelta(t1, t2):
    """A function to deal with image time intervals. Returns a number and
    time unit"""

    if t1 is None or t2 is None:
        logger.warning("One date is NULL ({0}, {1}) ignoring".format(t2, t1))
        return None, YEARS

    if t2 > t1:
        logger.warning("t2>t1 ({0}, {1}) ignoring".format(t2, t1))
        return None, YEARS

    # check for meaningful intervald
    if t1.year == 1900 or t2.year == 1900:
        logger.warning("Ignoring one date ({0}, {1})".format(t2, t1))
        return None, YEARS

    rdelta = relativedelta(t1, t2)

    if rdelta.years != 0:
        return rdelta.years, YEARS

    elif rdelta.months != 0:
        return rdelta.months, MONTHS

    else:
        return rdelta.days, DAYS


PATTERN_INTERVAL = re.compile(r"([\d]+) ([\w]+s?)")


def parse_image_timedelta(interval):
    """A function to parse from a image_timdelta string"""

    match = re.search(PATTERN_INTERVAL, interval)

    # get parsed data
    value, units = match.groups()

    # time units are plural in database
    if units[-1] != 's':
        units += 's'

    # get time units from database
    units = TIME_UNITS.get_value_by_desc(units)

    return int(value), units


# https://stackoverflow.com/a/39533619/4385116
# inspired django.contrib.admin.utils.get_deleted_objects, this function
# tries to determine all related objects starting from a provied one
# HINT: similar function at https://gist.github.com/nealtodd/4594575
def get_deleted_objects(objs, db_alias=DEFAULT_DB_ALIAS):
    # NestedObjects is an imporovement of django.db.models.deletion.Collector
    collector = NestedObjects(using=db_alias)
    collector.collect(objs)

    def format_callback(obj):
        opts = obj._meta
        no_edit_link = '%s: %s' % (capfirst(opts.verbose_name),
                                   force_text(obj))
        return no_edit_link

    to_delete = collector.nested(format_callback)
    protected = [format_callback(obj) for obj in collector.protected]
    model_count = {
        model._meta.verbose_name_plural:
            len(objs) for model, objs in collector.model_objs.items()}

    return to_delete, model_count, protected


async def send_message_to_websocket(message, pk):
    """
    Function will create websocket object and send message to django-channels

    Args:
        message (dict): message to send to websocket
        pk (str): primary key of submission
    """
    # Need to have it here as in case with small test data message sent to
    # websocket will overcome response from server
    time.sleep(3)
    async with websockets.connect(
            'ws://asgi:8001/image/ws/submissions/{}/'.format(pk)) as websocket:
        await websocket.send(json.dumps(message))


def format_attribute(value, terms=None, library_uri=OBO_URL, units=None):
    """Format a generic attribute into biosample dictionary"""

    if value is None:
        return None

    # pay attention to datetime objects
    if isinstance(value, datetime.date):
        value = str(value)

    # HINT: need I deal with multiple values?

    result = {}
    result["value"] = value

    if terms:
        result["terms"] = [{
            "url": "/".join([
                library_uri,
                terms])
        }]

    if units:
        result["units"] = units

    # return a list of dictionaries
    return [result]


def get_admin_emails():
    """Return admin email from image.settings"""

    ADMINS = settings.ADMINS

    # return all admin mail addresses
    return [admin[1] for admin in ADMINS]


def send_mail_to_admins(
        email_subject, email_message,
        default_from=settings.DEFAULT_FROM_EMAIL,
        addresses=get_admin_emails()):
    """Send a mail to all admins

    Args:
        email_subject (str): the email subject
        email_message (str): the body of the email
        default_from (str): the from field of the email
        addresses (list): a list of email addresses
    """

    # submit mail to admins
    datatuple = (
        email_subject,
        email_message,
        default_from,
        addresses)

    send_mass_mail((datatuple, ))


def uid2biosample(value):
    """Convert human-readable name to model field"""
    if value == 'Sample storage':
        return 'storage'
    elif value == 'Sample storage processing':
        return 'storage_processing'
    elif value == 'Sampling to preparation interval':
        return 'preparation_interval_units'
    elif value == 'Specimen collection protocol':
        return 'protocol'
    return '_'.join(value.lower().split(" "))


1			#!/usr/bin/env python3
2			# -- coding: utf-8 --
3			"""
4			Created on Wed Mar 27 12:50:16 2019
5
6			@author: Paolo Cozzi <[email protected]>
7			"""
8
9			import re
10			import logging
11			import datetime
12			import websockets
13			import time
14			import json
15
16			from dateutil.relativedelta import relativedelta
17
18			from django.conf import settings
19			from django.contrib.admin.utils import NestedObjects
20			from django.core.mail import send_mass_mail
21			from django.db import DEFAULT_DB_ALIAS
22			from django.utils.text import capfirst
23			from django.utils.encoding import force_text
24
25			from .constants import YEARS, MONTHS, DAYS, OBO_URL, TIME_UNITS
26
27			# Get an instance of a logger
28			logger = logging.getLogger(__name__)
29
30
31			def image_timedelta(t1, t2):
32			"""A function to deal with image time intervals. Returns a number and
33			time unit"""
34
35			if t1 is None or t2 is None:
36			logger.warning("One date is NULL ({0}, {1}) ignoring".format(t2, t1))
37			return None, YEARS
38
39			if t2 > t1:
40			logger.warning("t2>t1 ({0}, {1}) ignoring".format(t2, t1))
41			return None, YEARS
42
43			# check for meaningful intervald
44			if t1.year == 1900 or t2.year == 1900:
45			logger.warning("Ignoring one date ({0}, {1})".format(t2, t1))
46			return None, YEARS
47
48			rdelta = relativedelta(t1, t2)
49
50			if rdelta.years != 0:
51			return rdelta.years, YEARS
52
53			elif rdelta.months != 0:
54			return rdelta.months, MONTHS
55
56			else:
57			return rdelta.days, DAYS
58
59
60			PATTERN_INTERVAL = re.compile(r"([\d]+) ([\w]+s?)")
61
62
63			def parse_image_timedelta(interval):
64			"""A function to parse from a image_timdelta string"""
65
66			match = re.search(PATTERN_INTERVAL, interval)
67
68			# get parsed data
69			value, units = match.groups()
70
71			# time units are plural in database
72			if units[-1] != 's':
73			units += 's'
74
75			# get time units from database
76			units = TIME_UNITS.get_value_by_desc(units)
77
78			return int(value), units
79
80
81			# https://stackoverflow.com/a/39533619/4385116
82			# inspired django.contrib.admin.utils.get_deleted_objects, this function
83			# tries to determine all related objects starting from a provied one
84			# HINT: similar function at https://gist.github.com/nealtodd/4594575
85			def get_deleted_objects(objs, db_alias=DEFAULT_DB_ALIAS):
86			# NestedObjects is an imporovement of django.db.models.deletion.Collector
87			collector = NestedObjects(using=db_alias)
88			collector.collect(objs)
89
90			def format_callback(obj):
91			opts = obj._meta
92			no_edit_link = '%s: %s' % (capfirst(opts.verbose_name),
93			force_text(obj))
94			return no_edit_link
95
96			to_delete = collector.nested(format_callback)
97			protected = [format_callback(obj) for obj in collector.protected]
98			model_count = {
99			model._meta.verbose_name_plural:
100			len(objs) for model, objs in collector.model_objs.items()}
101
102			return to_delete, model_count, protected
103
104
105			async def send_message_to_websocket(message, pk):
106			"""
107			Function will create websocket object and send message to django-channels
108
109			Args:
110			message (dict): message to send to websocket
111			pk (str): primary key of submission
112			"""
113			# Need to have it here as in case with small test data message sent to
114			# websocket will overcome response from server
115			time.sleep(3)
116			async with websockets.connect(
117			'ws://asgi:8001/image/ws/submissions/{}/'.format(pk)) as websocket:
118			await websocket.send(json.dumps(message))
119
120
121			def format_attribute(value, terms=None, library_uri=OBO_URL, units=None):
122			"""Format a generic attribute into biosample dictionary"""
123
124			if value is None:
125			return None
126
127			# pay attention to datetime objects
128			if isinstance(value, datetime.date):
129			value = str(value)
130
131			# HINT: need I deal with multiple values?
132
133			result = {}
134			result["value"] = value
135
136			if terms:
137			result["terms"] = [{
138			"url": "/".join([
139			library_uri,
140			terms])
141			}]
142
143			if units:
144			result["units"] = units
145
146			# return a list of dictionaries
147			return [result]
148
149
150			def get_admin_emails():
151			"""Return admin email from image.settings"""
152
153			ADMINS = settings.ADMINS
154
155			# return all admin mail addresses
156			return [admin[1] for admin in ADMINS]
157
158
159			def send_mail_to_admins(
160			email_subject, email_message,
161			default_from=settings.DEFAULT_FROM_EMAIL,
162			addresses=get_admin_emails()):
163			"""Send a mail to all admins
164
165			Args:
166			email_subject (str): the email subject
167			email_message (str): the body of the email
168			default_from (str): the from field of the email
169			addresses (list): a list of email addresses
170			"""
171
172			# submit mail to admins
173			datatuple = (
174			email_subject,
175			email_message,
176			default_from,
177			addresses)
178
179			send_mass_mail((datatuple, ))
180
181
182			def uid2biosample(value):
183			"""Convert human-readable name to model field"""
184			if value == 'Sample storage':
185			return 'storage'
186			elif value == 'Sample storage processing':
187			return 'storage_processing'
188			elif value == 'Sampling to preparation interval':
189			return 'preparation_interval_units'
190			elif value == 'Specimen collection protocol':
191			return 'protocol'
192			return '_'.join(value.lower().split(" "))
193

cnr-ibba / IMAGE-InjectTool

Pull Request — master (#55)

common.helpers.send_mail_to_admins() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like