Skip to content

file

Download strategy class for the file scheme.

FileConfig

Bases: AttrDict

File-specific Configuration Data Model.

Source code in oteapi/strategies/download/file.py 
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: Optional[str] = Field(
        None,
        description=(
            "Encoding used when opening the file. The default is platform dependent."
        ),
    )
    datacache_config: Optional[DataCacheConfig] = Field(
        None,
        description="Configurations for the data cache for storing the downloaded file content.",
    )

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
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."
    )

    @validator("downloadUrl")
    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

ensure_path_exists(value)

Ensure path is defined in downloadUrl.

Source code in oteapi/strategies/download/file.py 
48
49
50
51
52
53
@validator("downloadUrl")
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 
62
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
@dataclass
class FileStrategy:
    """Strategy for retrieving data from a local file.

    **Registers strategies**:

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

    """

    download_config: FileResourceConfig

    def initialize(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdate:
        """Initialize."""
        return SessionUpdate()

    def get(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdateFile:
        """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 SessionUpdateFile(key=key)

get(session=None)

Read local file.

Source code in oteapi/strategies/download/file.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def get(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdateFile:
    """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 SessionUpdateFile(key=key)

initialize(session=None)

Initialize.

Source code in oteapi/strategies/download/file.py 
74
75
76
def initialize(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdate:
    """Initialize."""
    return SessionUpdate()

SessionUpdateFile

Bases: SessionUpdate

Class for returning values from Download File strategy.

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

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