Skip to content

openaleph_procrastinate.helpers

Helper functions to access Archive and FollowTheMoney data within Jobs

entity_writer(dataset)

Get the ftmstore.dataset.BulkLoader for the given dataset. The entities are flushed when leaving the context.

Source code in openaleph_procrastinate/helpers.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@contextmanager
def entity_writer(dataset: str) -> Generator[BulkLoader, None, None]:
    """
    Get the `ftmstore.dataset.BulkLoader` for the given `dataset`. The entities
    are flushed when leaving the context.
    """
    store = get_dataset(
        dataset, origin=OPAL_ORIGIN, database_uri=settings.ftm_store_uri
    )
    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.

Danger

This is not tested.

Source code in openaleph_procrastinate/helpers.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@contextmanager
def get_localpath(dataset: str, content_hash: str) -> Generator[Path, None, None]:
    """
    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.

    !!! danger
        This is not tested.
    """
    archive = get_archive()
    key = lookup_key(content_hash)
    store = get_virtual()
    path = store.download(key, archive)
    try:
        yield Path(path)
    finally:
        store.cleanup(path)

load_entity(dataset, entity_id)

Retrieve a single entity from the store.

Source code in openaleph_procrastinate/helpers.py 
59
60
61
62
63
64
65
66
67
def load_entity(dataset: str, entity_id: str) -> EntityProxy:
    """
    Retrieve a single entity from the store.
    """
    store = get_dataset(dataset, database_uri=settings.ftm_store_uri)
    entity = store.get(entity_id)
    if entity is None:
        raise EntityNotFound(f"Entity `{entity_id}` not found in dataset `{dataset}`")
    return entity

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.

Danger

This is not tested.

Source code in openaleph_procrastinate/helpers.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@contextmanager
def open_file(dataset: str, content_hash: str) -> Generator[BinaryIO, None, None]:
    """
    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.

    !!! danger
        This is not tested.
    """
    archive = get_archive()
    key = lookup_key(content_hash)
    with archive.open(key) as handler:
        try:
            yield handler
        finally:
            handler.close()