| 305 | |
| 306 | |
| 307 | class File(object): |
| 308 | _prefix_to_storage: dict = { |
| 309 | "oss": OSSStorage, |
| 310 | "http": HTTPStorage, |
| 311 | "https": HTTPStorage, |
| 312 | "local": LocalStorage, |
| 313 | } |
| 314 | |
| 315 | @staticmethod |
| 316 | def _get_storage(uri): |
| 317 | """Internal: get storage. |
| 318 | |
| 319 | Args: |
| 320 | uri: TODO. |
| 321 | """ |
| 322 | assert isinstance(uri, str), f"uri should be str type, but got {type(uri)}" |
| 323 | |
| 324 | if "://" not in uri: |
| 325 | # local path |
| 326 | storage_type = "local" |
| 327 | else: |
| 328 | prefix, _ = uri.split("://") |
| 329 | storage_type = prefix |
| 330 | |
| 331 | assert storage_type in File._prefix_to_storage, ( |
| 332 | f"Unsupported uri {uri}, valid prefixs: " f"{list(File._prefix_to_storage.keys())}" |
| 333 | ) |
| 334 | |
| 335 | if storage_type not in G_STORAGES: |
| 336 | G_STORAGES[storage_type] = File._prefix_to_storage[storage_type]() |
| 337 | |
| 338 | return G_STORAGES[storage_type] |
| 339 | |
| 340 | @staticmethod |
| 341 | def read(uri: str) -> bytes: |
| 342 | """Read data from a given ``filepath`` with 'rb' mode. |
| 343 | |
| 344 | Args: |
| 345 | filepath (str or Path): Path to read data. |
| 346 | |
| 347 | Returns: |
| 348 | bytes: Expected bytes object. |
| 349 | """ |
| 350 | storage = File._get_storage(uri) |
| 351 | return storage.read(uri) |
| 352 | |
| 353 | @staticmethod |
| 354 | def read_text(uri: Union[str, Path], encoding: str = "utf-8") -> str: |
| 355 | """Read data from a given ``filepath`` with 'r' mode. |
| 356 | |
| 357 | Args: |
| 358 | filepath (str or Path): Path to read data. |
| 359 | encoding (str): The encoding format used to open the ``filepath``. |
| 360 | Default: 'utf-8'. |
| 361 | |
| 362 | Returns: |
| 363 | str: Expected text reading from ``filepath``. |
| 364 | """ |
no outgoing calls
no test coverage detected
searching dependent graphs…