Source code for sewerrat.register

from typing import List, Optional, Union
import requests
import os
import warnings
import time

from . import _utils as ut




def register(path: str, names: Union[str, List[str]], url: str, retry: int = 3, wait: int = 1):
    """
    Register a directory into the SewerRat search index. It is assumed that
    that the directory is world-readable and that the caller has write access.
    If a metadata file cannot be indexed (e.g., due to incorrect formatting,
    insufficient permissions), a warning will be printed but the function will
    not throw an error.

    Args:
        path: 
            Path to the directory to be registered.

        names: 
            List of strings containing the base names of metadata files inside
            ``path`` to be indexed. Alternatively, a single string containing
            the base name for a single metadata file.

        url:
            URL to the SewerRat REST API.

        retry:
            Deprecated, ignored.

        wait:
            Deprecated, ignored.
    """
    if isinstance(names, str):
        names = [names]
    elif len(names) == 0:
        raise ValueError("expected at least one entry in 'names'")

    path = ut.clean_path(path)
    res = requests.post(url + "/register/start", json = { "path": path }, allow_redirects=True)
    if res.status_code >= 300:
        raise ut.format_error(res)

    body = res.json()
    code = body["code"]
    target = os.path.join(path, code)
    with open(target, "w") as handle:
        handle.write("")

    try:
        res = requests.post(url + "/register/finish", json = { "path": path, "base": names }, allow_redirects=True)
        if res.status_code >= 300:
            raise ut.format_error(res)
        body = res.json()
    finally:
        os.unlink(target)

    for comment in body["comments"]:
        warnings.warn(comment)
    return