mirror of
ssh://git.janware.com/janware/proj/jw-pkg
synced 2026-04-24 09:13:37 +02:00
Jan Lindemann
94eee5c4bb
Add the parameter "atomic" to put() / _put(). If instructs the implementation to take extra precautions to make sure the operation either succeeds or fails entirely, i.e. doesn't leave a broken target file behind. Signed-off-by: Jan Lindemann <jan@janware.com>
46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
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
|
|
|
|
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)
|