Skip to content

file

Download strategy class for the file scheme.

DownloadFileContent

Bases: AttrDict

Class for returning values from Download File strategy.

Source code in oteapi/strategies/download/file.py 
57
58
59
60
class DownloadFileContent(AttrDict):
    """Class for returning values from Download File strategy."""

    key: str = Field(..., description="Key to access the data in the cache.")

key = Field(..., description='Key to access the data in the cache.') class-attribute instance-attribute

FileConfig

Bases: AttrDict

File-specific Configuration Data Model.

Source code in oteapi/strategies/download/file.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class FileConfig(AttrDict):
    """File-specific Configuration Data Model."""

    text: bool = Field(
        False,
        description=(
            "Whether the file should be opened in text mode. If `False`, the file will"
            " be opened in bytes mode."
        ),
    )
    encoding: str | None = Field(
        None,
        description=(
            "Encoding used when opening the file. The default is platform dependent."
        ),
    )
    datacache_config: DataCacheConfig | None = Field(
        None,
        description=(
            "Configurations for the data cache for storing the downloaded file "
            "content."
        ),
    )

datacache_config = Field(None, description='Configurations for the data cache for storing the downloaded file content.') class-attribute instance-attribute

encoding = Field(None, description='Encoding used when opening the file. The default is platform dependent.') class-attribute instance-attribute

text = Field(False, description='Whether the file should be opened in text mode. If `False`, the file will be opened in bytes mode.') class-attribute instance-attribute

FileResourceConfig

Bases: ResourceConfig

File download strategy filter config.

Source code in oteapi/strategies/download/file.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class FileResourceConfig(ResourceConfig):
    """File download strategy filter config."""

    downloadUrl: FileUrl = Field(  # type: ignore[assignment]
        ..., description="The file URL, which will be downloaded."
    )
    configuration: FileConfig = Field(
        FileConfig(), description="File download strategy-specific configuration."
    )

    @field_validator("downloadUrl")
    @classmethod
    def ensure_path_exists(cls, value: FileUrl) -> FileUrl:
        """Ensure `path` is defined in `downloadUrl`."""
        if not value.path:
            raise ValueError("downloadUrl must contain a `path` part.")
        return value

configuration = Field(FileConfig(), description='File download strategy-specific configuration.') class-attribute instance-attribute

downloadUrl = Field(..., description='The file URL, which will be downloaded.') class-attribute instance-attribute

ensure_path_exists(value) classmethod

Ensure path is defined in downloadUrl.

Source code in oteapi/strategies/download/file.py 
48
49
50
51
52
53
54
@field_validator("downloadUrl")
@classmethod
def ensure_path_exists(cls, value: FileUrl) -> FileUrl:
    """Ensure `path` is defined in `downloadUrl`."""
    if not value.path:
        raise ValueError("downloadUrl must contain a `path` part.")
    return value

FileStrategy

Strategy for retrieving data from a local file.

Registers strategies:

  • ("scheme", "file")
Source code in oteapi/strategies/download/file.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@dataclass
class FileStrategy:
    """Strategy for retrieving data from a local file.

    **Registers strategies**:

    - `("scheme", "file")`

    """

    download_config: FileResourceConfig

    def initialize(self) -> AttrDict:
        """Initialize."""
        return AttrDict()

    def get(self) -> DownloadFileContent:
        """Read local file."""
        filename = uri_to_path(self.download_config.downloadUrl).resolve()

        if not filename.exists():
            raise FileNotFoundError(f"File not found at {filename}")

        cache = DataCache(self.download_config.configuration.datacache_config)
        if cache.config.accessKey and cache.config.accessKey in cache:
            key = cache.config.accessKey
        else:
            key = cache.add(
                filename.read_text(encoding=self.download_config.configuration.encoding)
                if self.download_config.configuration.text
                else filename.read_bytes()
            )

        return DownloadFileContent(key=key)

download_config instance-attribute

get()

Read local file.

Source code in oteapi/strategies/download/file.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def get(self) -> DownloadFileContent:
    """Read local file."""
    filename = uri_to_path(self.download_config.downloadUrl).resolve()

    if not filename.exists():
        raise FileNotFoundError(f"File not found at {filename}")

    cache = DataCache(self.download_config.configuration.datacache_config)
    if cache.config.accessKey and cache.config.accessKey in cache:
        key = cache.config.accessKey
    else:
        key = cache.add(
            filename.read_text(encoding=self.download_config.configuration.encoding)
            if self.download_config.configuration.text
            else filename.read_bytes()
        )

    return DownloadFileContent(key=key)

initialize()

Initialize.

Source code in oteapi/strategies/download/file.py 
75
76
77
def initialize(self) -> AttrDict:
    """Initialize."""
    return AttrDict()