lib.FileContext: Add URI manipulation methods

Add .schema_from_uri(), .split_uri(), .id() to define some
standardish way to dissect an URI the same way FileContext does.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-04-21 21:06:51 +02:00
commit 3360ec86cb
Signed by: jan
GPG key ID: 3750640C9E25DD61

View file

@ -4,6 +4,7 @@ from __future__ import annotations
import abc, re
from typing import TYPE_CHECKING
from functools import cached_property, cache
if TYPE_CHECKING:
from typing import Self
@ -26,6 +27,24 @@ class FileContext(abc.ABC):
async def __aexit__(self, exc_type, exc, tb):
await self.close()
@classmethod
def schema_from_uri(cls, uri: str) -> str:
tokens = re.split(r'://', uri)
return tokens[0] if tokens[0] != uri else 'file'
@classmethod
@cache
def split_uri(cls, uri: str) -> tuple[str, str]:
from urllib.parse import urlparse
p = urlparse(uri)
netloc = p.netloc if p.netloc else ''
return f'{cls.schema_from_uri(uri)}://{netloc}', p.path
@cached_property
def id(self) -> str:
id, path = self.split_uri(self.__uri)
return id
@property
def uri(self) -> str:
return self.__uri
@ -196,9 +215,7 @@ class FileContext(abc.ABC):
@classmethod
def create(cls, uri: str, *args, **kwargs) -> Self:
tokens = re.split(r'://', uri)
schema = tokens[0] if tokens[0] != uri else 'file'
match schema:
match cls.schema_from_uri(uri):
case 'local' | 'file':
from .ec.Local import Local
return Local(uri, *args, **kwargs)
@ -210,4 +227,4 @@ class FileContext(abc.ABC):
return Curl(uri, *args, **kwargs)
case _:
pass
raise Exception(f'Can\'t create file transfer instance for {uri} with unknown schema "{schema}"')
raise Exception(f'Can\'t create file transfer instance for {uri} with unknown schema "{cls.schema_from_uri(uri)}"')