Skip to content

file

Download strategy class for the file scheme.

FileConfig (BaseModel) pydantic-model

File-specific Configuration Data Model.

Source code in oteapi/strategies/download/file.py 
class FileConfig(BaseModel):
    """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."
        ),
    )

encoding: str pydantic-field

Encoding used when opening the file. The default is platform dependent.

text: bool pydantic-field

Whether the file should be opened in text mode. If False, the file will be opened in bytes mode.

FileStrategy dataclass

Strategy for retrieving data from a local file.

Registers strategies:

  • ("scheme", "file")
Source code in oteapi/strategies/download/file.py 
@dataclass
class FileStrategy:
    """Strategy for retrieving data from a local file.

    **Registers strategies**:

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

    """

    download_config: "ResourceConfig"

    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."""
        if (
            self.download_config.downloadUrl is None
            or self.download_config.downloadUrl.scheme != "file"
        ):
            raise ValueError(
                "Expected 'downloadUrl' to have scheme 'file' in the configuration."
            )

        filename = Path(self.download_config.downloadUrl.path).resolve()
        if isinstance(filename, PosixPath):
            filename = Path("/" + self.download_config.downloadUrl.host + str(filename))

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

        return SessionUpdateFile(key=key)

get(self, session=None)

Read local file.

Source code in oteapi/strategies/download/file.py 
def get(self, session: "Optional[Dict[str, Any]]" = None) -> SessionUpdateFile:
    """Read local file."""
    if (
        self.download_config.downloadUrl is None
        or self.download_config.downloadUrl.scheme != "file"
    ):
        raise ValueError(
            "Expected 'downloadUrl' to have scheme 'file' in the configuration."
        )

    filename = Path(self.download_config.downloadUrl.path).resolve()
    if isinstance(filename, PosixPath):
        filename = Path("/" + self.download_config.downloadUrl.host + str(filename))

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

    return SessionUpdateFile(key=key)

initialize(self, session=None)

Initialize.

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

SessionUpdateFile (SessionUpdate) pydantic-model

Class for returning values from Download File strategy.

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

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

key: str pydantic-field required

Key to access the data in the cache.

Back to top