lib + cmds.projects: Use lib.Uri

Remove the feeble attempts at unifying URI handling, and use class Uri from lib.Uri instead.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-04-27 09:37:28 +02:00
commit d9746cd20b
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
8 changed files with 65 additions and 98 deletions

View file

@ -3,26 +3,22 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from urllib.parse import urlparse
from ..FileContext import FileContext as Base
from ..base import Result
if TYPE_CHECKING:
from ..ExecContext import ExecContext
from ..Uri import Uri
class Curl(Base):
def __init__(self, uri: str, *args, ec: ExecContext|None=None, **kwargs) -> None:
def __init__(self, uri: str|Uri, *args, ec: ExecContext|None=None, **kwargs) -> None:
super().__init__(uri=uri, *args, **kwargs)
self.__ec: ExecContext|None = ec
if ec is None:
from .Local import Local
self.__ec = Local(interactive=False, *args, **kwargs)
p = urlparse(uri)
self.__base_url = f'{p.scheme}://{p.hostname}'
if p.port is not None:
self.__base_url += f':{p.port}'
async def _get(
self,
@ -42,5 +38,5 @@ class Curl(Base):
path = wd + '/' + path
if not len(path) or path[0] != '/':
path = '/' + path
cmd.append(self.__base_url + path)
cmd.append(self.url.to_string + self._chroot(path))
return await self.__ec.run(cmd, throw=throw, verbose=verbose)

View file

@ -1,5 +1,8 @@
# -*- coding: utf-8 -*-
from __future__ import annotations
from typing import TYPE_CHECKING
import os, sys, subprocess, asyncio, pwd, grp, stat
from functools import cache
@ -10,9 +13,12 @@ from ..base import Result, StatResult
from ..log import *
from ..util import pretty_cmd
if TYPE_CHECKING:
from ..Uri import Uri
class Local(Base):
def __init__(self, uri='local', *args, **kwargs) -> None:
def __init__(self, uri: str|Uri='local', *args, **kwargs) -> None:
super().__init__(uri, *args, **kwargs)
@cache

View file

@ -1,6 +1,8 @@
# -*- coding: utf-8 -*-
from typing import Any
from __future__ import annotations
from typing import Any, TYPE_CHECKING
import os, abc, sys, pwd
from enum import Flag, auto
@ -9,7 +11,10 @@ from ..util import pretty_cmd
from ..log import *
from ..base import Result
from ..ExecContext import ExecContext
from urllib.parse import urlparse
from ..Uri import Uri
if TYPE_CHECKING:
from ..Uri import Uri
class SSHClient(ExecContext):
@ -19,21 +24,12 @@ class SSHClient(ExecContext):
ModEnv = auto()
Wd = auto()
def __init__(self, uri: str, caps: Caps=Caps(0), *args, **kwargs) -> None:
def __init__(self, uri: Uri|str, caps: Caps=Caps(0), *args, **kwargs) -> None:
uri = Uri.pimp(uri)
if uri.username is None:
uri.set_username(pwd.getpwuid(os.getuid()).pw_name)
super().__init__(uri=uri, *args, **kwargs)
self.__caps = caps
try:
parsed = urlparse(uri)
except Exception as e:
log(ERR, f'Failed to parse SSH URI "{uri}"')
raise
self.__hostname = parsed.hostname
if self.__hostname is None:
raise Exception(f'Can\'t parse host name from SSH URI "{uri}"')
self.__port = parsed.port
self.__password = parsed.password
self.__username = parsed.username
@abc.abstractmethod
async def _run_ssh(
@ -105,26 +101,19 @@ class SSHClient(ExecContext):
@property
def hostname(self) -> str|None:
return self.__hostname
return self.uri.hostname
@property
def port(self) -> int|None:
return self.__port
return self.uri.port
def set_password(self, password: str) -> None:
self.__password = password
@property
def username(self) -> str|None:
return self.uri.username
@property
def password(self) -> str|None:
return self.__password
def set_username(self, username: str) -> None:
self.__username = username
def _username(self) -> str:
if self.__username is None:
return pwd.getpwuid(os.getuid()).pw_name
return self.__username
return self.uri.password
def ssh_client(*args, type: str|list[str]|None=None, **kwargs) -> SSHClient: # export
from importlib import import_module
@ -146,7 +135,7 @@ def ssh_client(*args, type: str|list[str]|None=None, **kwargs) -> SSHClient: # e
msg = f'Can\'t instantiate SSH client class {name} ({str(e)})'
errors.append(msg)
log(DEBUG, f'{msg}, trying next')
msg = f'No working SSH clients for {" ".join(args)}'
msg = f'No working SSH clients for {" ".join([str(arg) for arg in args])}'
log(ERR, f'----- {msg}')
for error in errors:
log(ERR, error)