Gitlab Community Edition Instance

Commit 15623165 authored by Marcel Hellkamp's avatar Marcel Hellkamp
Browse files

More prototyping

parent 5ad25cea
......@@ -17,24 +17,26 @@ class CDStar:
def __init__(self, url, auth=None, _session=None):
self.url = url.rstrip("/") + '/'
self.auth = auth
self.headers = {}
self._session = _session or requests.Session()
self._tx = None
self._autocommit = False
def clone(self):
""" Return an independant instance with the same settings.
""" Return an independent instance with the same settings.
Other state (e.g. running transaction) is not copied.
return CDStar(self.url, auth=self.auth, _session=self._session)
def _rest(self, method, *path, _expect_status=None, **options):
def _rest(self, method, *path, _expect_status=None, headers=None, **options):
if self.auth:
options['auth'] = self.auth
if self.headers:
options.setdefault("headers", {}).update(self.headers)
headers = headers or {}
if self.tx:
headers['X-Transaction'] = self.tx['id']
options['headers'] = headers
rs = self._session.request(
method, self.url + '/'.join(path), **options)
......@@ -44,109 +46,187 @@ class CDStar:
# TODO: handle errors
raise Exception(rs.text)
def tx(self):
return self._tx
def begin(self, autocommit=False, readonly=False):
""" Start a new transaction and return self.
def begin(self, autocommit=False, readonly=True):
if self._tx:
self._tx = TxHandle(self, autocommit=autocommit, readonly=readonly)
return self._tx
Transactions are used to group multiple operations into a single atomic action. After begin(), you have to
call commit() or rollback() to apply or undo all operations made while the transaction was active.
def service_info(self):
return self._rest("GET").json()
It is strongly recommended to only always start transactions as with-statements::
def vault_info(self, vault: str):
return self._rest("GET", vault).json()
with cdstar.begin():
# do stuff
def create_archive(self, vault, upload: "ComboUpdate" = None):
if upload:
return self._rest("POST", vault, data=upload.form,
headers={'Content-Type': upload.form.content_type}).json()
return self._rest("POST", vault).json()
You can commit() or rollback() early, or even begin() a new transaction while still in the with-block, but
there can only be a single transaction per client active at a time. On exit, the current transaction is
closed automatically. If autocommit is true and no exception was raised, it is committed. Otherwise, it is
rolled back.
def update_archive(self, vault, archive, upload: "ComboUpdate"):
return self._rest("POST", vault, archive, data=upload.form,
headers={'Content-Type': upload.form.content_type}).json()
:param autocommit: Commit this transaction automatically after the with-block without errors? (default: false)
:param readonly: Create a (cheap) read-only transaction.
self._autocommit = autocommit
self._tx = self._rest("POST", "_tx", data={"readonly": readonly}).json()
return self
def put_file(self, vault, id, target, source, type=None):
self._rest("PUT", vault, id, target.lstrip("/"),
headers={'Content-Type': type or "application/x-autodetect"}
def tx(self):
""" Return the current transaction handle, or None if no transaction is running. """
return self._tx
def commit(self) -> None:
""" Commit the current transaction. """
if not self._tx:
raise RuntimeError("No transaction running")
class TxHandle:
def __init__(self, client, autocommit=False, readonly=True):
self.client = client
self.autocommit = autocommit
self._tx = None
self._rest("POST", "_tx", self._tx['id'])
self._tx = None
except Exception:
def rollback(self) -> None:
""" Rollback the current transaction, if any. Do nothing otherwise. """
if self._tx:
self._rest("DELETE", "_tx", self._tx['id'])
self._tx = None
def keepalive(self):
""" If a transaction is running, keep it alive. Otherwise, do nothing. """
if not self._tx:
raise RuntimeError("No transaction running")
self._tx = self._rest("GET", "_tx", self._tx['id']).json()
def __enter__(self):
if self.client._tx:
self._tx = self.client._rest("POST", "_tx", data={"readonly": self.readonly}).json()
self._tx['x-autocommit'] = self.autocommit
self.client._tx = self
return self
""" Expect a transaction to be already running. """
if not self._tx:
raise RuntimeError("No transaction running. Call begin() frist.")
def __exit__(self, exc_type, exc_value, traceback):
if self._tx and self.client._tx is self:
if exc_type is None and self._tx['x-autocommit']:
""" Commit or roll-back the current transaction, depending on its autocommit setting. """
if self._tx:
if exc_type is None and self._autocommit:
self.client._tx = None
def id(self):
return self._tx['id']
def service_info(self):
return self._rest("GET").json()
def is_running(self):
return bool(self._tx)
def vault_info(self, vault: str):
return self._rest("GET", vault).json()
def renew(self):
if not self._tx:
raise RuntimeError("No transaction running")
self.client._rest("GET", "_tx", self._tx['id'])
def create_archive(self, vault, form: "FormUpdate" = None):
if form:
return self._rest("POST", vault, data=form.body,
headers={'Content-Type': form.content_type}).json()
return self._rest("POST", vault).json()
def commit(self):
if not self._tx:
raise RuntimeError("No transaction running")
self.client._rest("POST", "_tx", self._tx['id'])
self._tx = None
def update_archive(self, vault, archive, form: "FormUpdate"):
return self._rest("POST", vault, archive, data=form.body,
headers={'Content-Type': form.content_type}).json()
def rollback(self):
""" Rollback the current transaction.
def list_files(self, vault, archive, offset=0, limit=100):
""" Request a FileList for an archive.
Do nothing if no transaction is active.
The FileList may be incomplete of more than `limit` files are in an archive. See iter_files() for a
convenient way to get all files as an iterator.
if self._tx:
self.client._rest("DELETE", "_tx", self._tx['id'])
self._tx = None
query = {"files": "true", "offset": offset, "limit": limit}
return self._rest("GET", vault, archive, query=query).json()
def iter_files(self, vault, archive, offset=0, **args):
""" Yield all FileInfo entries of an archive.
This method may (lazily) issue more than one request if an archive contains more than `limit` files.
while True:
files = self.list_files(vault, archive, offset, **args)
if files['files'] and offset + files['count'] <= files['total']:
yield from files['files']
offset += files['count']
def put_file(self, vault, archive, name, source, type=None):
return self._rest("PUT", vault, archive, _fix_filename(name), data=source,
headers={'Content-Type': type or "application/x-autodetect"}).json()
def get_file(self, vault, archive, name):
""" Return a stream-able response object representing the requested file. """
return self._rest("GET", vault, archive, _fix_filename(name), stream=True)
def get_fileinfo(self, vault, archive, name, meta=True):
""" Get information about a file """
query = {"info": "true"}
if meta:
query['with'] = "meta"
return self._rest("GET", vault, archive, _fix_filename(name), query=query).json()
def search(self, vault, q, order=None, limit=0, scroll=None, groups=None):
""" Perform a search and return the SearchResults document.
See iter_search() for a way to fetch more than `limit` results.
query = {"q": q}
if order:
query['order'] = order
if limit:
query['limit'] = limit
if scroll:
query['scroll'] = scroll
if groups:
query['groups'] = groups
return self._rest("GET", vault, query=query).json()
def iter_search(self, vault, q, scroll=None, **args):
""" Yield all SearchHit entries of a search.
This method may (lazily) issue more than one request if a search returns more than `limit` results.
while True:
hits =, q, scroll=scroll or "", **args)
if hits['hits']:
yield from hits['hits']
scroll = hits['scroll']
def _fix_filename(name):
# silently strip leading slashes
name = name.lstrip("/")
# Fail hard on relative filenames
if name != os.path.normpath(name):
raise ValueError( "Archive file name not in a normalized form: {} != {}".format(name, os.path.normpath(name)))
return name
class PostUpdate:
class FormUpdate:
""" Builder for CDSTAR POST multipart/form-data requests to upload multiple files or change aspects of an archive.
def __init__(self):
self.fields = []
self._form = None
self._mp = None
def form(self):
if not self._form:
self._form = MultipartEncoder(self.fields)
return self._form
def body(self):
if not self._mp:
self._mp = MultipartEncoder(self.fields)
return self._mp
def content_type(self):
return self.body.content_type
def upload(self, target, src, type="application/x-autodetect"):
""" Upload a file (string path or opened file-like object)
:param target: Target within the archive (must start with '/')
:param src: String path to an existing file, or opened file-like object.
:param type: Mime-type of the upload.
......@@ -3,9 +3,9 @@ Upload one or more files to an existing archive.
import os
from ..context import VerboseTx
from .. import register_subcommand, printer, CliError
from .._utils import compile_glob, hbytes
from .._utils import compile_glob, hbytes
def register():
parser = register_subcommand("put", command, help=__doc__.splitlines()[0], description=__doc__)
......@@ -18,7 +18,7 @@ def register():
help="Exclude files by glob pattern")
parser.add_argument("-i", "--include", metavar="GLOB", action="append",
help="Include files by glob pattern (default: all)")
parser.add_argument("-p", "--progress", action="store_true",
parser.add_argument("-%", "--progress", action="store_true",
help="Show progress bar for large files or slow uploads")
parser.add_argument("TARGET", help="Archive ID, or 'new' to create a new archive")
......@@ -86,7 +86,7 @@ def command(ctx, args):
len(uploads), hbytes(total), "new archive" if archive == 'new' else "archive: " + vault + "/" + archive)
with VerboseTx(client.begin(autocommit=True)) as tx:
with client.begin(autocommit=True):
if archive == 'new':
archive = client.create_archive(vault)['id']
printer("Uploading {} files ({}) to new archive: {}/{}", len(uploads), hbytes(total), vault, archive)
......@@ -104,7 +104,7 @@ def command(ctx, args):
for i, target in enumerate(sorted(uploads)):
file, stat = uploads[target]
with open(file, 'rb') as fp:
line = "[{}/{}] {}".format(i + 1, len(uploads), target[1:])
line = "[{}/{}] {} ({})".format(i + 1, len(uploads), target[1:], hbytes(stat.st_size))
if pbar:
read =
......@@ -118,4 +118,5 @@ def command(ctx, args):
if pbar:
printer("Upload complete!")
printer("Uploaded {} files ({}) to archive: {}/{}", len(uploads), hbytes(total), vault, archive)
......@@ -97,22 +97,6 @@ class CliContext:
raise RuntimeError("Asking for password not implemented yet")
class VerboseTx:
def __init__(self, tx):
self.tx = tx
def __enter__(self):
tx = self.tx.__enter__()
printer.v("Transaction started: {}",
return tx
def __exit__(self, exc_type, exc_val, exc_tb):
id =
self.tx.__exit__(exc_type, exc_val, exc_tb)
if exc_type:
printer("Transaction rolled back: {}", id)
class ArchiveDir:
""" A folder containing a `.cdstar` archive metadata directory. """
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment