2620

pending completion

Build # 2620

Build Type

Pull #779

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/CenterForOpenScience/SHARE/commit/<a class=hub.com/CenterForOpenScience/SHARE/commit/<a class="double-link" href="https://git"><a class=hub.com/CenterForOpenScience/SHARE/commit/<a class="double-link" href="https://git"><a class=hub.com/CenterForOpenScience/SHARE/commit/cd1707ab510a0ea95b11768007dd9e28e3cea946">cd1707ab5<a href="https://github.com/CenterForOpenScience/SHARE/commit/cd1707ab510a0ea95b11768007dd9e28e3cea946">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/CenterForOpenScience/SHARE/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/CenterForOpenScience/SHARE/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/CenterForOpenScience/SHARE/commit/cd1707ab510a0ea95b11768007dd9e28e3cea946&quot;&gt;cd1707ab5&lt;/a&gt;&lt;a href=&quot;https://github.com/CenterForOpenScience/SHARE/commit/cd1707ab510a0ea95b11768007dd9e28e3cea946&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/CenterForOpenScience/SHARE/commit/cd1707ab510a0ea95b11768007dd9e28&lt;/a&gt;e3cea946&quot;>&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/CenterForOpenScience/SHARE/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/CenterForOpenScience/SHARE/commit/<a class="double-link" href="https://github.com/CenterForOpenScience/SHARE/commit/9f9502a43cce398bec8f32f329dfaf2d434eaeda">9f9502a43</a><a href="https://github.com/CenterForOpenScience/SHARE/commit/cd1707ab510a0ea95b11768007dd9e28e3cea946">&amp;quot;&amp;gt;9f9502a43&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;9f9502a43&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/CenterForOpenScience/SHARE/commit/cd1707ab510a0ea95b11768007dd9e28e3cea&lt;/a&gt;946&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/CenterForOpenScience/SHARE/commit/<a class="double-link" href="https://github.com/CenterForOpenScience/SHARE/commit/9c776fb4d">9c776fb4d">9c776fb4d</a>

Pull Request Pull Request #779: [wip][ENG-1987] indexer shenanigans

Run Details

598 of 598 new or added lines in 30 files covered. (100.0%)

10201 of 12855 relevant lines covered (79.35%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.76

/share/ingest/ingester.py

import json
import pendulum
from typing import Union

from share.exceptions import IngestError
from share.ingest.scheduler import IngestScheduler
from share.models import RawDatum
from share.models import SourceConfig
from share.tasks import ingest
from share.tasks.jobs import IngestJobConsumer


class Ingester:
    """Helper class that takes a datum and feeds it to SHARE

    Usage given a source config:
        Ingester(datum, datum_id).with_config(source_config).ingest()

    Usage given a user and transformer:
        Ingester(datum, datum_id).as_user(user, 'transformer_key').ingest()
    """

    raw = None
    job = None
    async_task = None

    _config = None

    def __init__(self, datum: Union[str, list, dict], datum_id: str, datestamp=None):
        if isinstance(datum, str):
            self.datum = datum
        elif isinstance(datum, (list, dict)):
            self.datum = json.dumps(datum, sort_keys=True)
        else:
            raise TypeError('datum must be a string or a json-serializable dict or list')

        if not datum_id:
            raise IngestError('Ingester requires a datum_id (for suid\'s sake)')

        self.datum_id = datum_id
        self.datestamp = datestamp or pendulum.now()

    def with_config(self, config):
        assert not self._config
        self._config = config
        return self

    def as_user(self, user, transformer_key='v2_push'):
        """Ingest as the given user, with the given transformer

        Create a source config for the given user/transformer, or get a previously created one.
        """
        assert not self._config
        self._config = SourceConfig.objects.get_or_create_push_config(user, transformer_key)
        return self

    def ingest(self, **kwargs):
        # "Here comes the airplane!"
        assert 'job_id' not in kwargs
        self._setup_ingest(claim_job=True)
        IngestJobConsumer().consume(job_id=self.job.id, exhaust=False, **kwargs)
        return self

    def ingest_async(self, start_task=True, **kwargs):
        # "There's pizza in the fridge."
        assert 'job_id' not in kwargs
        self._setup_ingest(claim_job=start_task)
        if start_task:
            self.async_task = ingest.delay(job_id=self.job.id, exhaust=False, **kwargs)
        return self

    def _setup_ingest(self, claim_job):
        assert self.datum and self._config and not (self.raw or self.job or self.async_task)

        # TODO get rid of FetchResult, or make it more sensical
        from share.harvest.base import FetchResult
        fetch_result = FetchResult(self.datum_id, self.datum, self.datestamp)
        self.raw = RawDatum.objects.store_data(self._config, fetch_result)
        self.job = IngestScheduler().schedule(self.raw.suid, self.raw.id, claim=claim_job)

1	import json	1✔
2	import pendulum	1✔
3	from typing import Union	1✔
4
5	from share.exceptions import IngestError	1✔
6	from share.ingest.scheduler import IngestScheduler	1✔
7	from share.models import RawDatum	1✔
8	from share.models import SourceConfig	1✔
9	from share.tasks import ingest	1✔
10	from share.tasks.jobs import IngestJobConsumer	1✔
11
12
13	class Ingester:	1✔
14	"""Helper class that takes a datum and feeds it to SHARE
15
16	Usage given a source config:
17	Ingester(datum, datum_id).with_config(source_config).ingest()
18
19	Usage given a user and transformer:
20	Ingester(datum, datum_id).as_user(user, 'transformer_key').ingest()
21	"""
22
23	raw = None	1✔
24	job = None	1✔
25	async_task = None	1✔
26
27	_config = None	1✔
28
29	def __init__(self, datum: Union[str, list, dict], datum_id: str, datestamp=None):	1✔
30	if isinstance(datum, str):	1✔
31	self.datum = datum	×
32	elif isinstance(datum, (list, dict)):	1✔
33	self.datum = json.dumps(datum, sort_keys=True)	1✔
34	else:
35	raise TypeError('datum must be a string or a json-serializable dict or list')	×
36
37	if not datum_id:	1✔
38	raise IngestError('Ingester requires a datum_id (for suid\'s sake)')	×
39
40	self.datum_id = datum_id	1✔
41	self.datestamp = datestamp or pendulum.now()	1✔
42
43	def with_config(self, config):	1✔
44	assert not self._config	×
45	self._config = config	×
46	return self	×
47
48	def as_user(self, user, transformer_key='v2_push'):	1✔
49	"""Ingest as the given user, with the given transformer
50
51	Create a source config for the given user/transformer, or get a previously created one.
52	"""
53	assert not self._config	1✔
54	self._config = SourceConfig.objects.get_or_create_push_config(user, transformer_key)	1✔
55	return self	1✔
56
57	def ingest(self, **kwargs):	1✔
58	# "Here comes the airplane!"
59	assert 'job_id' not in kwargs	1✔
60	self._setup_ingest(claim_job=True)	1✔
61	IngestJobConsumer().consume(job_id=self.job.id, exhaust=False, **kwargs)	1✔
62	return self	1✔
63
64	def ingest_async(self, start_task=True, **kwargs):	1✔
65	# "There's pizza in the fridge."
66	assert 'job_id' not in kwargs	1✔
67	self._setup_ingest(claim_job=start_task)	1✔
68	if start_task:	1✔
69	self.async_task = ingest.delay(job_id=self.job.id, exhaust=False, **kwargs)	1✔
70	return self	1✔
71
72	def _setup_ingest(self, claim_job):	1✔
73	assert self.datum and self._config and not (self.raw or self.job or self.async_task)	1✔
74
75	# TODO get rid of FetchResult, or make it more sensical
76	from share.harvest.base import FetchResult	1✔
77	fetch_result = FetchResult(self.datum_id, self.datum, self.datestamp)	1✔
78	self.raw = RawDatum.objects.store_data(self._config, fetch_result)	1✔
79	self.job = IngestScheduler().schedule(self.raw.suid, self.raw.id, claim=claim_job)	1✔

CenterForOpenScience / SHARE / 2620

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous