Skip to content

openaleph_procrastinate.helpers

Helper functions to access Archive and FollowTheMoney data within Jobs

entity_writer(dataset, origin=OPAL_ORIGIN)

Get the ftmq.store.fragments.BulkLoader for the given dataset. The writer is flushed when leaving the context.

Source code in openaleph_procrastinate/helpers.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@contextmanager
def entity_writer(
    dataset: str, origin: str = OPAL_ORIGIN
) -> Generator[BulkLoader, None, None]:
    """
    Get the `ftmq.store.fragments.BulkLoader` for the given `dataset`. The
    writer is flushed when leaving the context.
    """
    store = get_fragments(
        dataset,
        origin=origin,
        database_uri=settings.fragments_uri,
        **sqlalchemy_pool,
    )
    loader = store.bulk()
    try:
        yield loader
    finally:
        loader.flush()

get_localpath(dataset, content_hash)

Load a file from the archive and store it in a local temporary path for further processing. The file is cleaned up after leaving the context. Reference

Source code in openaleph_procrastinate/helpers.py 
26
27
28
29
30
31
32
33
34
def get_localpath(dataset: str, content_hash: str) -> ContextManager[Path]:
    """
    Load a file from the archive and store it in a local temporary path for
    further processing. The file is cleaned up after leaving the context.
    [Reference][openaleph_procrastinate.model.DatasetJob.get_file_references]
    """
    archive = get_archive()
    key = lookup_key(content_hash)
    return archive.local_path(key)

load_entities(dataset, entity_ids)

Batch retrieve entities from the fragment store.

Source code in openaleph_procrastinate/helpers.py 
49
50
51
52
53
54
55
56
57
58
def load_entities(
    dataset: str, entity_ids: Iterable[str]
) -> Generator[EntityProxy, None, None]:
    """
    Batch retrieve entities from the fragment store.
    """
    store = get_fragments(
        dataset, database_uri=settings.fragments_uri, **sqlalchemy_pool
    )
    yield from store.iterate(entity_ids)

open_file(dataset, content_hash)

Load a file from the archive and store it in a local temporary path for further processing. Returns an open file handler. The file is closed and cleaned up after leaving the context. Reference

Source code in openaleph_procrastinate/helpers.py 
37
38
39
40
41
42
43
44
45
46
def open_file(dataset: str, content_hash: str) -> ContextManager[VirtualIO]:
    """
    Load a file from the archive and store it in a local temporary path for
    further processing. Returns an open file handler. The file is closed and
    cleaned up after leaving the context.
    [Reference][openaleph_procrastinate.model.DatasetJob.get_file_references]
    """
    archive = get_archive()
    key = lookup_key(content_hash)
    return archive.local_open(key, algorithm=ARCHIVE_CHECKSUM_ALGORITHM)