lib.ec.Curl: Add class

Add class Curl as the first pure FileTransfer class without _run()
/ _sudo(). It doesn't use any PycURL / libcurl-like binding, but runs
the curl binary in a subprocess. That looks the most portable still.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-04-16 12:03:58 +02:00
commit 45001144d7

View file

@ -0,0 +1,60 @@
# -*- coding: utf-8 -*-
from __future__ import annotations
from typing import TYPE_CHECKING
from urllib.parse import urlparse
from ..FileTransfer import FileTransfer as Base
from ..base import Result
if TYPE_CHECKING:
from ..ExecContext import ExecContext
class Curl(Base):
def __init__(self, uri: str, *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,
path: str,
wd: str|None,
throw: bool,
verbose: bool|None,
title: str
) -> Result:
cmd = ['curl']
if verbose is None:
verbose = self.__ec.verbose_default
if not verbose:
cmd.append('-s')
cmd.append('--follow')
if wd is not None:
path = wd + '/' + path
if not len(path) or path[0] != '/':
path = '/' + path
cmd.append(self.__base_url + path)
return await self.__ec.run(cmd, throw=throw, verbose=verbose)
async def _put(
self,
content: bytes,
path: str,
wd: str|None,
throw: bool,
verbose: bool|None,
title: str,
owner: str|None,
group: str|None,
mode: str|None,
) -> Result:
raise NotImplementedError()