huanld/pygeoapi
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

add links, refactor job handling, HTML updates (#599)

Browse Source 
* add links, refactor job handling, HTML updates

* fix test

* misc fixes
This commit is contained in:
Tom Kralidis
2020-12-30 09:30:12 -05:00
committed by GitHub
parent e481891f7c
commit 6915efbcad
7 changed files with 143 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pygeoapi/api.py
+93 -135
View File
@@ -1903,13 +1903,21 @@ tiles/{{{}}}/{{{}}}/{{{}}}/{{{}}}?f=mvt'
link = {
'type': 'text/html',
'rel': 'collection',
'href': jobs_url,
'title': 'Collection of jobs for the {} process'.format(
key),
'href': '{}?f=html'.format(jobs_url),
'title': 'jobs for this process as HTML',
'hreflang': self.config['server'].get('language', None)
}
p2['links'].append(link)
link = {
'type': 'application/json',
'rel': 'collection',
'href': '{}?f=json'.format(jobs_url),
'title': 'jobs for this process as JSON',
'hreflang': self.config['server'].get('language', None)
}
p2['links'].append(link)
processes.append(p2)
if process is not None:
@@ -1932,13 +1940,14 @@ tiles/{{{}}}/{{{}}}/{{{}}}/{{{}}}?f=mvt'
return headers_, 200, to_json(response, self.pretty_print)
def get_process_jobs(self, headers, args, process_id):
def get_process_jobs(self, headers, args, process_id, job_id=None):
"""
Get process jobs
:param headers: dict of HTTP headers
:param args: dict of HTTP request parameters
:param process_id: id of process
:param job_id: id of job
:returns: tuple of headers, status code, content
"""
@@ -1971,26 +1980,77 @@ tiles/{{{}}}/{{{}}}/{{{}}}/{{{}}}?f=mvt'
p = load_plugin('process', processes[process_id]['processor'])
if self.manager:
jobs = sorted(self.manager.get_jobs(process_id),
key=lambda k: k['process_start_datetime'],
reverse=True)
if job_id is None:
jobs = sorted(self.manager.get_jobs(process_id),
key=lambda k: k['job_start_datetime'],
reverse=True)
else:
jobs = [self.manager.get_job(process_id, job_id)]
else:
LOGGER.debug('Process management not configured')
jobs = []
serialized_jobs = []
for job_ in jobs:
job2 = {
'jobID': job_['identifier'],
'status': job_['status'],
'message': job_['message'],
'progress': job_['progress'],
'job_start_datetime': job_['job_start_datetime'],
'job_end_datetime': job_['job_end_datetime']
}
if JobStatus[job_['status']] in [
JobStatus.successful, JobStatus.running, JobStatus.accepted]:
job_result_url = '{}/processes/{}/jobs/{}/results'.format(
self.config['server']['url'],
process_id, job_['identifier'])
job2['links'] = [{
'href': '{}?f=html'.format(job_result_url),
'rel': 'about',
'type': 'text/html',
'title': 'results of job {} as HTML'.format(job_id)
}, {
'href': '{}?f=json'.format(job_result_url),
'rel': 'about',
'type': 'application/json',
'title': 'results of job {} as JSON'.format(job_id)
}]
if job_['mimetype'] not in ['application/json', 'text/html']:
job2['links'].append({
'href': job_result_url,
'rel': 'about',
'type': job_['mimetype'],
'title': 'results of job {} as {}'.format(
job_id, job_['mimetype'])
})
serialized_jobs.append(job2)
if job_id is None:
j2_template = 'jobs.html'
else:
serialized_jobs = serialized_jobs[0]
j2_template = 'job.html'
if format_ == 'html':
headers_['Content-Type'] = 'text/html'
data = {
'process': {'id': process_id, 'title': p.metadata['title']},
'jobs': jobs,
'process': {
'id': process_id,
'title': p.metadata['title']
},
'jobs': serialized_jobs,
'now': datetime.now(timezone.utc).strftime(DATETIME_FORMAT)
}
response = render_j2_template(self.config, 'jobs.html', data)
response = render_j2_template(self.config, j2_template, data)
return headers_, 200, response
response = [job['identifier'] for job in jobs]
return headers_, 200, to_json(response, self.pretty_print)
return headers_, 200, to_json(serialized_jobs, self.pretty_print)
def execute_process(self, headers, args, data, process_id):
"""
@@ -2138,111 +2198,6 @@ tiles/{{{}}}/{{{}}}/{{{}}}/{{{}}}?f=mvt'
return headers_, http_status, to_json(response, self.pretty_print)
def get_process_job(self, headers, args, process_id, job_id):
"""
Get status of job (instance of a process)
:param headers: dict of HTTP headers
:param args: dict of HTTP request parameters
:param process_id: process identifier
:param job_id: job identifier
:returns: tuple of headers, status code, content
"""
headers_ = HEADERS.copy()
processes = filter_dict_by_key_value(self.config['resources'],
'type', 'process')
if process_id not in processes:
exception = {
'code': 'NoSuchProcess',
'description': 'identifier not found'
}
LOGGER.info(exception)
return headers_, 404, to_json(exception, self.pretty_print)
job = self.manager.get_job(process_id, job_id)
if not job:
exception = {
'code': 'NoSuchJob',
'description': 'job not found'
}
LOGGER.info(exception)
return headers_, 404, to_json(exception, self.pretty_print)
format_ = check_format(args, headers)
if format_ is not None and format_ not in FORMATS:
exception = {
'code': 'InvalidParameterValue',
'description': 'Invalid format'
}
LOGGER.error(exception)
return headers_, 400, to_json(exception, self.pretty_print)
status = JobStatus[job['status']]
response = {
'jobID': job_id,
'status': status.value,
'message': job.get('message', None),
'links': [{
'href': '{}/processes/{}/jobs/{}'.format(
self.config['server']['url'], process_id, job_id
),
'rel': 'self',
'type': 'application/json',
'title': 'Status of {} job {}'.format(process_id, job_id)
}]
}
if status in (JobStatus.successful, JobStatus.running,
JobStatus.accepted):
# TODO link also if accepted/running?
response['links'].append({
'href': '{}/processes/{}/jobs/{}/results'.format(
self.config['server']['url'], process_id, job_id
),
'rel': 'about',
'type': 'application/json',
'title': 'Results of {} job {}'.format(process_id, job_id)
})
elif status == JobStatus.failed:
# TODO link to exception report?
pass
if format_ != 'html':
return headers_, 200, json.dumps(response, default=json_serial)
else:
headers_['Content-Type'] = 'text/html'
process = load_plugin('process',
processes[process_id]['processor'])
process_info = {
'id': process_id,
'title': process.metadata['title']
}
psd = job.get('process_start_datetime', None)
ped = job.get('process_end_datetime', None)
if status == JobStatus.successful:
progress = 100
else:
progress = job.get('progress', 0)
job_info = {
'process_start_datetime': psd,
'process_end_datetime': ped,
'progress': progress,
**response
}
return headers_, 200, render_j2_template(self.config, 'job.html', {
'process': process_info,
'job': job_info,
'now': datetime.now(timezone.utc).strftime(DATETIME_FORMAT),
})
def get_process_job_result(self, headers, args, process_id, job_id):
"""
Get result of job (instance of a process)
@@ -2313,25 +2268,28 @@ tiles/{{{}}}/{{{}}}/{{{}}}/{{{}}}?f=mvt'
LOGGER.info(exception)
return headers_, 400, json.dumps(exception)
job_output = self.manager.get_job_result(process_id, job_id)
mimetype, job_output = self.manager.get_job_result(process_id, job_id)
format_ = check_format(args, headers)
if format_ == 'html':
headers_['Content-Type'] = 'text/html'
data = {
'process': {
'id': process_id, 'title': process.metadata['title']
},
'job': {'id': job_id},
'result': job_output
}
response = render_j2_template(self.config, 'job_result.html',
data)
return headers_, 200, response
content = json.dumps(job_output, sort_keys=True, indent=4,
default=json_serial)
if mimetype not in [None, 'application/json']:
headers_['Content-Type'] = mimetype
content = job_output
else:
if format_ == 'json':
content = json.dumps(job_output, sort_keys=True, indent=4,
default=json_serial)
else:
headers_['Content-Type'] = 'text/html'
data = {
'process': {
'id': process_id, 'title': process.metadata['title']
},
'job': {'id': job_id},
'result': job_output
}
content = render_j2_template(self.config, 'job_result.html',
data)
return headers_, 200, content
pygeoapi/flask_app.py
+1 -1
View File
@@ -404,7 +404,7 @@ def get_process_jobs(process_id=None, job_id=None):
headers, status_code, content = api_.delete_process_job(
process_id, job_id)
else: # Return status of a specific job
headers, status_code, content = api_.get_process_job(
headers, status_code, content = api_.get_process_jobs(
request.headers, request.args, process_id, job_id)
response = make_response(content, status_code)
pygeoapi/process/manager/base.py
+18 -11
View File
@@ -112,7 +112,7 @@ class BaseManager:
:param process_id: process identifier
:param job_id: job identifier
:returns: `str` of raw output or None
:returns: `tuple` of mimetype and raw output
"""
raise NotImplementedError()
@@ -165,23 +165,18 @@ class BaseManager:
:returns: tuple of response payload and status
"""
if self.output_dir is not None:
filename = '{}-{}'.format(p.metadata['id'], job_id)
job_filename = os.path.join(self.output_dir, filename)
else:
job_filename = 'stdout'
process_id = p.metadata['id']
current_status = JobStatus.accepted
job_metadata = {
'identifier': job_id,
'process_id': process_id,
'process_start_datetime': datetime.utcnow().strftime(
'job_start_datetime': datetime.utcnow().strftime(
DATETIME_FORMAT),
'process_end_datetime': None,
'job_end_datetime': None,
'status': current_status.value,
'location': None,
'mimetype': None,
'message': 'Job accepted and ready for execution',
'progress': 5
}
@@ -189,8 +184,17 @@ class BaseManager:
self.add_job(job_metadata)
try:
if self.output_dir is not None:
filename = '{}-{}'.format(p.metadata['id'], job_id)
job_filename = os.path.join(self.output_dir, filename)
else:
job_filename = None
jfmt = p.metadata['outputs'][0]['output']['formats'][0]['mimeType']
current_status = JobStatus.running
outputs = p.execute(data_dict)
self.update_job(process_id, job_id, {
'status': current_status.value,
'message': 'Writing job output',
@@ -203,11 +207,13 @@ class BaseManager:
fh.write(json.dumps(outputs, sort_keys=True, indent=4))
current_status = JobStatus.successful
job_update_metadata = {
'process_end_datetime': datetime.utcnow().strftime(
'job_end_datetime': datetime.utcnow().strftime(
DATETIME_FORMAT),
'status': current_status.value,
'location': job_filename,
'mimetype': jfmt,
'message': 'Job complete',
'progress': 100
}
@@ -231,10 +237,11 @@ class BaseManager:
}
LOGGER.error(err)
job_metadata = {
'process_end_datetime': datetime.utcnow().strftime(
'job_end_datetime': datetime.utcnow().strftime(
DATETIME_FORMAT),
'status': current_status.value,
'location': None,
'mimetype': None,
'message': f'{code}: {outputs["description"]}'
}
pygeoapi/process/manager/tinydb_.py
+8 -4
View File
@@ -179,26 +179,30 @@ class TinyDBManager(BaseManager):
:param process_id: process identifier
:param jobid: job identifier
:returns: The process output as a `dict`
:returns: `tuple` of mimetype and raw output
"""
job_result = self.get_job(process_id, job_id)
if not job_result:
# processs/job does not exist
return None
location = job_result.get('location', None)
mimetype = job_result.get('mimetype', None)
job_status = JobStatus[job_result['status']]
if not job_status == JobStatus.successful:
# Job is incomplete
return None
return (None,)
if not location:
# Job data was not written for some reason
# TODO log/raise exception?
return {}
return (None,)
with io.open(location, 'r', encoding='utf-8') as filehandler:
result = json.load(filehandler)
return result
return mimetype, result
def __repr__(self):
return '<TinyDBManager> {}'.format(self.name)
pygeoapi/templates/job.html
+18 -23
View File
@@ -2,9 +2,9 @@
{% block title %}{{ super() }} Job status {% endblock %}
{% block crumbs %}{{ super() }}
/ <a href="../../">Processes</a>
/ <a href="../">{{ data.process.title }}</a>
/ <a href="../">{{ data['process']['title'] }}</a>
/ <a href="./">Jobs</a>
/ <a href="./{{ data.job.jobID }}">{{ data.job.jobID }}</a>
/ <a href="./{{ data['jobs']['jobID'] }}">{{ data['jobs']['jobID'] }}</a>
{% endblock %}
{% block body %}
<section id="job">
@@ -13,43 +13,38 @@
</div>
<div class="row">
<div class="col-md-12">
<div id="job-status" class="{{ data.job.status }}">
<div id="job-status-header" class="{{ data.job.status }}">
<p>Status: {{ data['job']['status'] }}</p>
<p class="progress">Progress: {{ data.job.progress }}%</p>
<div id="job-status" class="{{ data['jobs']['status'] }}">
<div id="job-status-header" class="{{ data['jobs']['status'] }}">
<p>Status: {{ data['jobs']['status'] }}</p>
<p class="progress">Progress: {{ data['jobs']['progress'] }}%</p>
</div>
<div id="job-status-body">
<div class="message">
<h3>Message</h3>
<p>{{ data.job.message }}</p>
<p>{{ data['jobs']['message'] }}</p>
</div>
<div class="duration">
<h4><label for="progress">Progress</label></h4>
<progress id="progress" class="inline" value="{{data.job.progress|int*10}}" max="1000"></progress>
<progress id="progress" class="inline" value="{{ data['jobs']['progress']|int*10 }}" max="1000"></progress>
<h4><label for="runtime">Duration</label></h4>
<p><span id="runtime">
{% if data.job.status == 'running' %}
{{ data.job.process_start_datetime|format_duration(data.now) }}
{% if data['jobs']['status'] == 'running' %}
{{ data['jobs']['job_start_datetime']|format_duration(data.now) }}
{% else %}
{{ data.job.process_start_datetime|format_duration(data.job.process_end_datetime) }}
{{ data['jobs']['job_start_datetime']|format_duration(data['jobs']['job_end_datetime']) }}
{% endif %}
</span></p>
<h4><label for="starttime">Started processing</label></h4>
<p><span id="starttime">{{ data.job.process_start_datetime|format_datetime }}</span></p>
<p><span id="starttime">{{ data['jobs']['job_start_datetime']|format_datetime }}</span></p>
<h4><label for="endtime">Finished processing</label></h4>
<p><span id="endtime">{{ data.job.process_end_datetime|format_datetime }}</span></p>
</div>
<div class="links">
{% if data.job.links %}
<h4><label for="resources">Resources</label></h4>
<ul id="resources">
{% for link in data.job.links %}
{% if link.rel != 'self' %}
<li><a href="{{ link.href }}" rel="{{ link.rel }}" type="{{ link.type }}" title="{{ link.title }}">{{ link.title }}</a></li>
{% endif %}
<p><span id="endtime">{{ data['jobs']['job_end_datetime']|format_datetime }}</span></p>
<h3>Links</h3>
<ul>
{% for link in data['jobs']['links'] %}
<li><a title="{{ link['rel'] }}" href="{{ link['href'] }}"><span>{{ link['title'] }}</span> (<span>{{ link['type'] }}</span>)</a></li>
{% endfor %}
</ul>
{% endif %}
</div>
</div>
</div>
pygeoapi/templates/jobs.html
+4 -9
View File
@@ -5,9 +5,6 @@
/ <a href="../{{ data.process.id }}">{{ data.process.title }}</a>
/ <a href="./jobs">Jobs</a>
{% endblock %}
{% block form_js %}
<script src="{{ config['server']['url'] }}/static/js/deleteJob.js"></script>
{% endblock %}
{% block body %}
<section id="jobs">
<div class="row">
@@ -27,15 +24,13 @@
<tbody>
{% for job in data.jobs %}
<tr>
<td class="small"><a href="/processes/{{data.process.id}}/jobs/{{ job.identifier }}">{{ job.identifier }}</a></td>
<td>
{{ job.process_start_datetime|format_datetime }}
</td>
<td class="small"><a href="{{ config['server']['url'] }}/processes/{{data.process.id}}/jobs/{{ job.jobID}}">{{ job.jobID }}</a></td>
<td><abbr title="{{ job.job_start_datetime|format_datetime }}">{{ job.job_start_datetime|format_datetime }}</abbr></td>
<td>
{% if job.status == 'running' %}
{{ job.process_start_datetime|format_duration(data.now) }}
{{ job.job_start_datetime|format_duration(data.now) }}
{% else %}
{{ job.process_start_datetime|format_duration(job.process_end_datetime) }}
{{ job.job_start_datetime|format_duration(job.job_end_datetime) }}
{% endif %}
</td>
<td>
tests/test_api.py
+1 -1
View File
@@ -758,7 +758,7 @@ def test_describe_processes(config, api_):
assert process['version'] == '0.2.0'
assert process['title'] == 'Hello World'
assert len(process['keywords']) == 3
assert len(process['links']) == 2
assert len(process['links']) == 3
assert len(process['inputs']) == 2
assert len(process['outputs']) == 1
assert len(process['outputTransmission']) == 1