Skip to content

Source

Source

Bases: BaseModel

A model describing an arbitrary local or remote source.

Source code in investigraph/model/source.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Source(BaseModel):
    """
    A model describing an arbitrary local or remote source.
    """

    name: str
    """Identifier of the source (defaults to slugified uri)"""

    uri: str
    """Local or remote uri of this source (via `anystore` / `fsspec`)"""

    scheme: str
    """Uri scheme, is set automatically during initialization"""

    pandas: Playbook | None = None
    """Pandas transformation spec (via `runpandarun`)"""

    data: dict | None = {}
    """Arbitrary extra data"""

    def __init__(self, **data):
        data["uri"] = str(data["uri"])
        data["name"] = data.get("name", slugify(data["uri"]))
        data["scheme"] = data.get("scheme", urlparse(data["uri"]).scheme or "file")
        super().__init__(**data)

    def ensure_uri(self, base: PathLike) -> None:
        """
        ensure absolute file paths based on base path of parent config.yml
        """
        uri = self.uri
        if self.is_local:
            uri = absolute_path(uri, base)
        self.uri = ensure_uri(uri)

    @anycache(
        store=get_store("memory://"), key_func=lambda self: self.uri, model=SourceInfo
    )
    def info(self) -> SourceInfo:
        store, uri = get_store_for_uri(self.uri)
        return SourceInfo(**store.info(uri).model_dump())

    @property
    def is_local(self) -> bool:
        return self.scheme == SCHEME_FILE

    @property
    def mimetype(self) -> str:
        return self.info().mimetype

data = {} class-attribute instance-attribute

Arbitrary extra data

name instance-attribute

Identifier of the source (defaults to slugified uri)

pandas = None class-attribute instance-attribute

Pandas transformation spec (via runpandarun)

scheme instance-attribute

Uri scheme, is set automatically during initialization

uri instance-attribute

Local or remote uri of this source (via anystore / fsspec)

ensure_uri(base)

ensure absolute file paths based on base path of parent config.yml

Source code in investigraph/model/source.py 
55
56
57
58
59
60
61
62
def ensure_uri(self, base: PathLike) -> None:
    """
    ensure absolute file paths based on base path of parent config.yml
    """
    uri = self.uri
    if self.is_local:
        uri = absolute_path(uri, base)
    self.uri = ensure_uri(uri)