mirror of
https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git
synced 2025-06-22 05:50:34 +00:00
Change directory structure (move ./nopaque/* to ./)
This commit is contained in:
Patrick Jentsch
54
app/tasks/__init__.py
Normal file
54
app/tasks/__init__.py
Normal file
@ -0,0 +1,54 @@
|
||||
from .corpus_utils import CheckCorporaMixin
|
||||
from .job_utils import CheckJobsMixin
|
||||
from ..import db, socketio
|
||||
import docker
|
||||
|
||||
|
||||
class TaskRunner(CheckCorporaMixin, CheckJobsMixin):
|
||||
def __init__(self):
|
||||
self.docker = docker.from_env()
|
||||
self._socketio_message_buffer = {}
|
||||
|
||||
def run(self):
|
||||
self.check_corpora()
|
||||
self.check_jobs()
|
||||
db.session.commit()
|
||||
self.flush_socketio_messages()
|
||||
|
||||
def buffer_socketio_message(self, event, payload, room,
|
||||
msg_id=None, override_policy='replace'):
|
||||
if room not in self._socketio_message_buffer:
|
||||
self._socketio_message_buffer[room] = {}
|
||||
if event not in self._socketio_message_buffer[room]:
|
||||
self._socketio_message_buffer[room][event] = {}
|
||||
if msg_id is None:
|
||||
msg_id = len(self._socketio_message_buffer[room][event].keys())
|
||||
if override_policy == 'append':
|
||||
if msg_id in self._socketio_message_buffer[room][event]:
|
||||
# If the current message value isn't a list, convert it!
|
||||
if not isinstance(self._socketio_message_buffer[room][event][msg_id], list): # noqa
|
||||
self._socketio_message_buffer[room][event][msg_id] = [self._socketio_message_buffer[room][event][msg_id]] # noqa
|
||||
else:
|
||||
self._socketio_message_buffer[room][event][msg_id] = []
|
||||
self._socketio_message_buffer[room][event][msg_id].append(payload)
|
||||
elif override_policy == 'replace':
|
||||
self._socketio_message_buffer[room][event][msg_id] = payload
|
||||
else:
|
||||
raise Exception('Unknown override policy: {}'.format(override_policy)) # noqa
|
||||
return msg_id
|
||||
|
||||
def buffer_user_patch_operation(self, ressource, patch_operation):
|
||||
self.buffer_socketio_message('user_{}_patch'.format(ressource.user_id),
|
||||
patch_operation,
|
||||
'user_{}'.format(ressource.user_id),
|
||||
msg_id='_', override_policy='append')
|
||||
|
||||
def clear_socketio_message_buffer(self):
|
||||
self._socketio_message_buffer = {}
|
||||
|
||||
def flush_socketio_messages(self):
|
||||
for room in self._socketio_message_buffer:
|
||||
for event in self._socketio_message_buffer[room]:
|
||||
for message in self._socketio_message_buffer[room][event]:
|
||||
socketio.emit(event, self._socketio_message_buffer[room][event][message], room=room) # noqa
|
||||
self.clear_socketio_message_buffer()
|
||||
231
app/tasks/corpus_utils.py
Normal file
231
app/tasks/corpus_utils.py
Normal file
@ -0,0 +1,231 @@
|
||||
from ..models import Corpus
|
||||
import docker
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
|
||||
|
||||
class CheckCorporaMixin:
|
||||
def check_corpora(self):
|
||||
corpora = Corpus.query.all()
|
||||
queued_corpora = list(filter(lambda corpus: corpus.status == 'queued', corpora)) # noqa
|
||||
running_corpora = list(filter(lambda corpus: corpus.status == 'running', corpora)) # noqa
|
||||
start_analysis_corpora = list(filter(lambda corpus: corpus.status == 'start analysis', corpora)) # noqa
|
||||
analysing_corpora = list(filter(lambda corpus: corpus.status == 'analysing', corpora)) # noqa
|
||||
stop_analysis_corpora = list(filter(lambda corpus: corpus.status == 'stop analysis', corpora)) # noqa
|
||||
submitted_corpora = list(filter(lambda corpus: corpus.status == 'submitted', corpora)) # noqa
|
||||
for corpus in submitted_corpora:
|
||||
self.create_build_corpus_service(corpus)
|
||||
for corpus in queued_corpora + running_corpora:
|
||||
self.checkout_build_corpus_service(corpus)
|
||||
for corpus in start_analysis_corpora:
|
||||
self.create_cqpserver_container(corpus)
|
||||
for corpus in analysing_corpora:
|
||||
self.checkout_analysing_corpus_container(corpus)
|
||||
for corpus in stop_analysis_corpora:
|
||||
self.remove_cqpserver_container(corpus)
|
||||
|
||||
def create_build_corpus_service(self, corpus):
|
||||
corpus_data_dir = os.path.join(corpus.path, 'data')
|
||||
shutil.rmtree(corpus_data_dir, ignore_errors=True)
|
||||
os.mkdir(corpus_data_dir)
|
||||
corpus_registry_dir = os.path.join(corpus.path, 'registry')
|
||||
shutil.rmtree(corpus_registry_dir, ignore_errors=True)
|
||||
os.mkdir(corpus_registry_dir)
|
||||
corpus_file = os.path.join(corpus.path, 'merged', 'corpus.vrt')
|
||||
service_kwargs = {
|
||||
'command': 'docker-entrypoint.sh build-corpus',
|
||||
'constraints': ['node.role==worker'],
|
||||
'labels': {'origin': 'nopaque',
|
||||
'type': 'corpus.build',
|
||||
'corpus_id': str(corpus.id)},
|
||||
'mounts': [corpus_file + ':/root/files/corpus.vrt:ro',
|
||||
corpus_data_dir + ':/corpora/data:rw',
|
||||
corpus_registry_dir + ':/usr/local/share/cwb/registry:rw'],
|
||||
'name': 'build-corpus_{}'.format(corpus.id),
|
||||
'restart_policy': docker.types.RestartPolicy()
|
||||
}
|
||||
service_image = \
|
||||
'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'
|
||||
try:
|
||||
self.docker.services.create(service_image, **service_kwargs)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Create "{}" service raised '.format(service_kwargs['name'])
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
else:
|
||||
corpus.status = 'queued'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
|
||||
def checkout_build_corpus_service(self, corpus):
|
||||
service_name = 'build-corpus_{}'.format(corpus.id)
|
||||
try:
|
||||
service = self.docker.services.get(service_name)
|
||||
except docker.errors.NotFound:
|
||||
logging.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.NotFound" The service does not exist. '
|
||||
+ '(corpus.status: {} -> failed)'.format(corpus.status)
|
||||
)
|
||||
corpus.status = 'failed'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
except docker.errors.InvalidVersion:
|
||||
logging.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.InvalidVersion" One of the arguments is '
|
||||
+ 'not supported with the current API version.'
|
||||
)
|
||||
else:
|
||||
service_tasks = service.tasks()
|
||||
if not service_tasks:
|
||||
return
|
||||
task_state = service_tasks[0].get('Status').get('State')
|
||||
if corpus.status == 'queued' and task_state != 'pending':
|
||||
corpus.status = 'running'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
elif (corpus.status == 'running'
|
||||
and task_state in ['complete', 'failed']):
|
||||
try:
|
||||
service.remove()
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Remove "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
corpus.status = 'prepared' if task_state == 'complete' \
|
||||
else 'failed'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
|
||||
def create_cqpserver_container(self, corpus):
|
||||
corpus_data_dir = os.path.join(corpus.path, 'data')
|
||||
corpus_registry_dir = os.path.join(corpus.path, 'registry')
|
||||
container_kwargs = {
|
||||
'command': 'cqpserver',
|
||||
'detach': True,
|
||||
'volumes': [corpus_data_dir + ':/corpora/data:rw',
|
||||
corpus_registry_dir + ':/usr/local/share/cwb/registry:rw'],
|
||||
'name': 'cqpserver_{}'.format(corpus.id),
|
||||
'network': 'nopaque_default'
|
||||
}
|
||||
container_image = \
|
||||
'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'
|
||||
# Check if a cqpserver container already exists. If this is the case,
|
||||
# remove it and create a new one
|
||||
try:
|
||||
container = self.docker.containers.get(container_kwargs['name'])
|
||||
except docker.errors.NotFound:
|
||||
pass
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Get "{}" container raised '.format(container_kwargs['name'])
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
try:
|
||||
container.remove(force=True)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Remove "{}" container raised '.format(
|
||||
container_kwargs['name'])
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
try:
|
||||
self.docker.containers.run(container_image, **container_kwargs)
|
||||
except docker.errors.ContainerError:
|
||||
# This case should not occur, because detach is True.
|
||||
logging.error(
|
||||
'Run "{}" container raised '.format(container_kwargs['name'])
|
||||
+ '"docker.errors.ContainerError" The container exits with a '
|
||||
+ 'non-zero exit code and detach is False.'
|
||||
)
|
||||
corpus.status = 'failed'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
except docker.errors.ImageNotFound:
|
||||
logging.error(
|
||||
'Run "{}" container raised '.format(container_kwargs['name'])
|
||||
+ '"docker.errors.ImageNotFound" The specified image does not '
|
||||
+ 'exist.'
|
||||
)
|
||||
corpus.status = 'failed'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Run "{}" container raised '.format(container_kwargs['name'])
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
else:
|
||||
corpus.status = 'analysing'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
|
||||
def checkout_analysing_corpus_container(self, corpus):
|
||||
container_name = 'cqpserver_{}'.format(corpus.id)
|
||||
try:
|
||||
self.docker.containers.get(container_name)
|
||||
except docker.errors.NotFound:
|
||||
logging.error('Could not find "{}" but the corpus state is "analysing".') # noqa
|
||||
corpus.status = 'prepared'
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Get "{}" container raised '.format(container_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
|
||||
def remove_cqpserver_container(self, corpus):
|
||||
container_name = 'cqpserver_{}'.format(corpus.id)
|
||||
try:
|
||||
container = self.docker.containers.get(container_name)
|
||||
except docker.errors.NotFound:
|
||||
pass
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Get "{}" container raised '.format(container_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
try:
|
||||
container.remove(force=True)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Remove "{}" container raised '.format(container_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
corpus.status = 'prepared'
|
||||
patch_operation = {
|
||||
'op': 'replace', 'path': '/corpora/{}/status'.format(corpus.id), 'value': corpus.status}
|
||||
self.buffer_user_patch_operation(corpus, patch_operation)
|
||||
206
app/tasks/job_utils.py
Normal file
206
app/tasks/job_utils.py
Normal file
@ -0,0 +1,206 @@
|
||||
from datetime import datetime
|
||||
from werkzeug.utils import secure_filename
|
||||
from .. import db, mail
|
||||
from ..email import create_message
|
||||
from ..models import Job, JobResult
|
||||
import docker
|
||||
import logging
|
||||
import json
|
||||
import os
|
||||
|
||||
|
||||
# TODO: Integrate the service_settings into app/services/__init__.py
|
||||
service_settings = {
|
||||
'file-setup': {
|
||||
'default_args': ' --mem-mb 2048 --n-cores 2',
|
||||
'ressources': docker.types.Resources(cpu_reservation=2 * (10 ** 9),
|
||||
mem_reservation=2048 * (10 ** 6))
|
||||
},
|
||||
'nlp': {
|
||||
'default_args': ' --mem-mb 2048 --n-cores 2',
|
||||
'ressources': docker.types.Resources(cpu_reservation=2 * (10 ** 9),
|
||||
mem_reservation=2048 * (10 ** 6))
|
||||
},
|
||||
'ocr': {
|
||||
'default_args': ' --mem-mb 4096 --n-cores 4',
|
||||
'ressources': docker.types.Resources(cpu_reservation=4 * (10 ** 9),
|
||||
mem_reservation=4096 * (10 ** 6))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class CheckJobsMixin:
|
||||
def check_jobs(self):
|
||||
jobs = Job.query.all()
|
||||
canceling_jobs = list(filter(lambda job: job.status == 'canceling', jobs)) # noqa
|
||||
queued_jobs = list(filter(lambda job: job.status == 'queued', jobs))
|
||||
running_jobs = list(filter(lambda job: job.status == 'running', jobs))
|
||||
submitted_jobs = list(filter(lambda job: job.status == 'submitted', jobs)) # noqa
|
||||
for job in submitted_jobs:
|
||||
self.create_job_service(job)
|
||||
for job in queued_jobs + running_jobs:
|
||||
self.checkout_job_service(job)
|
||||
for job in canceling_jobs:
|
||||
self.remove_job_service(job)
|
||||
|
||||
def create_job_service(self, job):
|
||||
cmd = job.service
|
||||
cmd += ' -i /input'
|
||||
cmd += ' -o /output'
|
||||
cmd += ' --log-dir /input'
|
||||
cmd += ' --zip [{}]_{}'.format(job.service, secure_filename(job.title))
|
||||
cmd += service_settings[job.service]['default_args']
|
||||
cmd += ' ' + ' '.join(json.loads(job.service_args))
|
||||
# Setup input mount
|
||||
input_mount_src = job.path
|
||||
input_mount_dest = os.path.abspath('/input')
|
||||
if job.service == 'file-setup':
|
||||
input_mount_dest = os.path.join(input_mount_dest, secure_filename(job.title)) # noqa
|
||||
input_mount = '{}:{}:rw'.format(input_mount_src, input_mount_dest)
|
||||
# Setup output mount
|
||||
output_mount_src = os.path.join(job.path, 'output')
|
||||
output_mount_dest = os.path.abspath('/output')
|
||||
os.makedirs(output_mount_src)
|
||||
output_mount = '{}:{}:rw'.format(output_mount_src, output_mount_dest)
|
||||
service_kwargs = {'command': cmd,
|
||||
'constraints': ['node.role==worker'],
|
||||
'labels': {'origin': 'nopaque',
|
||||
'type': 'job',
|
||||
'job_id': str(job.id)},
|
||||
'mounts': [input_mount, output_mount],
|
||||
'name': 'job_{}'.format(job.id),
|
||||
'resources': service_settings[job.service]['ressources'], # noqa
|
||||
'restart_policy': docker.types.RestartPolicy()}
|
||||
service_image = 'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/{}:{}'.format(job.service, job.service_version) # noqa
|
||||
try:
|
||||
self.docker.services.create(service_image, **service_kwargs)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Create "{}" service raised '.format(service_kwargs['name'])
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
job.status = 'queued'
|
||||
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
|
||||
self.buffer_user_patch_operation(job, patch_operation)
|
||||
finally:
|
||||
self.send_job_notification(job)
|
||||
|
||||
def checkout_job_service(self, job):
|
||||
service_name = 'job_{}'.format(job.id)
|
||||
try:
|
||||
service = self.docker.services.get(service_name)
|
||||
except docker.errors.NotFound:
|
||||
logging.error('Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.NotFound" The service does not exist. '
|
||||
+ '(job.status: {} -> failed)'.format(job.status))
|
||||
job.status = 'failed'
|
||||
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
|
||||
self.buffer_user_patch_operation(job, patch_operation)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
except docker.errors.InvalidVersion:
|
||||
logging.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.InvalidVersion" One of the arguments is '
|
||||
+ 'not supported with the current API version.'
|
||||
)
|
||||
return
|
||||
else:
|
||||
service_tasks = service.tasks()
|
||||
if not service_tasks:
|
||||
return
|
||||
task_state = service_tasks[0].get('Status').get('State')
|
||||
if job.status == 'queued' and task_state != 'pending':
|
||||
job.status = 'running'
|
||||
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
|
||||
self.buffer_user_patch_operation(job, patch_operation)
|
||||
elif job.status == 'running' and task_state in ['complete', 'failed']:
|
||||
try:
|
||||
service.remove()
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Remove "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. ' # noqa
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
else:
|
||||
if task_state == 'complete':
|
||||
results_dir = os.path.join(job.path, 'output')
|
||||
result_files = filter(lambda x: x.endswith('.zip'),
|
||||
os.listdir(results_dir))
|
||||
for result_file in result_files:
|
||||
job_result = JobResult(filename=result_file, job=job) # noqa
|
||||
db.session.add(job_result)
|
||||
db.session.flush()
|
||||
db.session.refresh(job_result)
|
||||
patch_operation = {'op': 'add', 'path': '/jobs/{}/results/{}'.format(job.id, job_result.id), 'value': job_result.to_dict()} # noqa
|
||||
self.buffer_user_patch_operation(job, patch_operation) # noqa
|
||||
job.end_date = datetime.utcnow()
|
||||
patch_operation = {'op': 'replace', 'path': '/jobs/{}/end_date'.format(job.id), 'value': job.end_date.timestamp()} # noqa
|
||||
self.buffer_user_patch_operation(job, patch_operation)
|
||||
job.status = task_state
|
||||
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
|
||||
self.buffer_user_patch_operation(job, patch_operation)
|
||||
finally:
|
||||
self.send_job_notification(job)
|
||||
|
||||
def remove_job_service(self, job):
|
||||
service_name = 'job_{}'.format(job.id)
|
||||
try:
|
||||
service = self.docker.services.get(service_name)
|
||||
except docker.errors.NotFound:
|
||||
job.status = 'canceled'
|
||||
patch_operation = {'op': 'replace', 'path': '/jobs/{}/status'.format(job.id), 'value': job.status} # noqa
|
||||
self.buffer_user_patch_operation(job, patch_operation)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
except docker.errors.InvalidVersion:
|
||||
logging.error(
|
||||
'Get "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.InvalidVersion" One of the arguments is '
|
||||
+ 'not supported with the current API version.'
|
||||
)
|
||||
return
|
||||
else:
|
||||
try:
|
||||
service.update(mounts=None)
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Update "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
return
|
||||
try:
|
||||
service.remove()
|
||||
except docker.errors.APIError as e:
|
||||
logging.error(
|
||||
'Remove "{}" service raised '.format(service_name)
|
||||
+ '"docker.errors.APIError" The server returned an error. '
|
||||
+ 'Details: {}'.format(e)
|
||||
)
|
||||
|
||||
def send_job_notification(self, job):
|
||||
if job.creator.setting_job_status_mail_notifications == 'none':
|
||||
return
|
||||
if (job.creator.setting_job_status_mail_notifications == 'end'
|
||||
and job.status not in ['complete', 'failed']):
|
||||
return
|
||||
msg = create_message(job.creator.email,
|
||||
'Status update for your Job "{}"'.format(job.title), # noqa
|
||||
'tasks/email/notification', job=job)
|
||||
mail.send(msg)
|
||||
Reference in New Issue
Block a user