Skip to content

LiquidBT API

LiquidBT's main entrypoint.

main(*args, *, plugins=[], **kwargs)

The build system runtime.

For the non-keyword arguments, pass a PackageConfig instance per each package. For the plugins argument, pass a list of plugins to use.

Source code in liquidbt/__init__.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def main(
    *args, plugins: List[Plugin] = [], **kwargs,
):
    """
    The build system runtime.

    For the non-keyword arguments, pass a PackageConfig instance
        per each package.
    For the `plugins` argument, pass a list of plugins to use.
    """

    parser = argparse.ArgumentParser(
        description="The CLI for your build system."
    )
    parser.add_argument(
        "command", type=str, nargs="?", help="the command you want to run."
    )

    bar = CustomProgressBar().post_init()

    command = parser.parse_args().command
    if command is None:
        command = "build"

    files = kwargs.get("files", [])
    packages = kwargs.get("packages", [])

    if packages == [] and files == [] and plugins == []:
        raise RuntimeError(
            "You need to add build configs or plugins to the main() arguments!"
        )

    ctx = RunContext(command, bar)

    print("\n")

    if command == "build":
        for f in files:
            ctx.add_task(_create_build_task(ctx, f, False))

        for package in packages:
            ctx.add_task(_create_build_task(ctx, package, True))

    for plugin in plugins:
        plugin.load(ctx)

    while True:
        for task in ctx.get_tasks():
            if task.status is TaskStatuses.READY:
                ctx._progress_bar.before_task(task)
                task()
                ctx._progress_bar.after_task()
                task.status = TaskStatuses.COMPLETED
                continue
        break

    for plugin in plugins:
        plugin.shutdown()

    bar.do_output("✓ Done!")
    bar.update()

    bar.finish()

build_tools special

api

handle_package(ctx, package)

Builds a full package.

Source code in liquidbt/build_tools/api.py
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def handle_package(ctx: RunContext, package):
    """Builds a full package."""

    setuptoolsargs = package.setuptools_args
    pkgname = package.pkgname

    # create all the temp files
    unsafely_clean(pkgname, False)
    create_or_clear("tmpsetup.py")
    shutil.copytree(f"{pkgname}.s", pkgname)
    write_setup_file(ctx, setuptoolsargs, pkgname)

    stringbuilder = ""

    if len(package.formats) < 1:
        raise RuntimeError("No formats specified!")

    for format in package.formats:
        stringbuilder += f" {str(format)}"

    for file in os.listdir(pkgname):
        actualfile = "/".join([pkgname, file])
        code = open(actualfile, "r").read()

        # clear the file
        handler = open(actualfile, "w")
        # have the transformers do their thing
        for transformer in ctx.get_build_transformers():
            e = transformer(code)
            if e is not None and type(e) == str:
                code = e
        handler.write(code)
        handler.close()

    setuptools_launch_wrapper(stringbuilder)

    if os.getenv("DEBUG_NO_CLEAN_ON_END") is None:
        unsafely_clean(pkgname, package.keepsrc)

handle_single_file(ctx, file)

Runs transformation actions for a single file.

Source code in liquidbt/build_tools/api.py
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def handle_single_file(ctx: RunContext, file):
    """Runs transformation actions for a single file."""

    # the contents of the source file
    code = open(file, "r").read()

    # clear the file which will hold the transformed code
    handler = open(file.replace(".py", "") + "_dist.py", "w")

    # have the transformers do their thing
    for transformer in ctx.get_build_transformers():
        e = transformer(code)
        if e is not None and type(e) == str:
            code = e
    handler.write(code)
    handler.close()

handlers

Utilities.

create_or_clear(file)

Creates a file if it doesn't already exist, otherwise clears any existing content from it.

Source code in liquidbt/build_tools/handlers.py
11
12
13
14
15
16
17
18
19
20
def create_or_clear(file: str):
    """
    Creates a file if it doesn't already exist,
    otherwise clears any existing content from it.
    """

    try:
        open(file, mode="w")
    except FileExistsError:
        os.remove(file)

setuptools_launch_wrapper(setuptools_args)

Launch setuptools with the given arguments.

Source code in liquidbt/build_tools/handlers.py
49
50
51
52
53
54
55
def setuptools_launch_wrapper(setuptools_args: str):
    """Launch setuptools with the given arguments."""

    Popen(
        sys.executable + " tmpsetup.py --quiet" + setuptools_args,
        shell=True
    ).wait()

unsafely_clean(pkgname, keepsrc)

Cleans up the files left behind for the named package.

Source code in liquidbt/build_tools/handlers.py
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def unsafely_clean(pkgname: str, keepsrc: bool):
    """Cleans up the files left behind for the named package."""

    try:
        os.remove("tmpsetup.py")
    except FileNotFoundError:
        pass
    if not keepsrc:
        try:
            shutil.rmtree(pkgname)
        except Exception:
            pass
    try:
        shutil.rmtree(f"{pkgname}.egg-info")
    except Exception:
        pass

write_setup_file(ctx, setuptoolsargs, pkgname)

Writes the setup file.

Source code in liquidbt/build_tools/handlers.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def write_setup_file(ctx: RunContext, setuptoolsargs, pkgname: str):
    """Writes the setup file."""

    with open("tmpsetup.py", mode="a") as fh:
        # start off with basic data
        fh.write("import setuptools\nsetuptools.setup(\n")
        fh.write(f'\n    name="{pkgname}",')

        # these options are required
        setuptoolsargs["zip_safe"] = False
        setuptoolsargs["include_package_data"] = True

        for key in setuptoolsargs:
            val = setuptoolsargs[key]
            if type(val) == list or type(val) == dict or type(val) == bool:
                # doesn't need string inclosing after the =
                fh.write(f"\n    {key}={tostring(val)},")
            else:
                # does need string inclosing after the =
                fh.write(f'\n    {key}="{tostring(val)}",')

        # write the final metadata
        fh.write(f'\n    packages=["{pkgname}"]')
        fh.write("\n)")

typeClasses

DistFormat

An abstract distribution format.

PackageConfig

The configuration for a single package.

__init__(self, **kwargs) special

Creates the class.

Source code in liquidbt/build_tools/typeClasses.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def __init__(self, **kwargs):
    """Creates the class."""

    self.pkgname = kwargs["name"]
    self.kwargs = kwargs
    self.keepsrc = False
    self.formats = kwargs.get("formats", [])

    assert self.pkgname is not None

    if kwargs.get("keep_generated_sources", False):
        self.keepsrc = True

    self.setuptools_args = self._iter_args(kwargs)
add_format(self, d)

Adds the passed DistFormat.

Source code in liquidbt/build_tools/typeClasses.py
47
48
49
50
51
52
def add_format(self, d):
    """Adds the passed DistFormat."""

    if not isinstance(d, DistFormat):
        raise TypeError("Incorrect format specified!")
    self.formats.append(d)

SourceDist

A source distribution, typically a .tar.gz.

WheelBinaryDist

A wheel distribution (.whl file).

plugins

The plugin API.

Plugin

Typical plugin interface.

name: str property readonly

The plugin's name.

__init__(self, *args, **kwargs) special

Called on initialization of the plugin class.

Source code in liquidbt/plugins.py
 7
 8
 9
10
11
def __init__(self, *args, **kwargs):
    """Called on initialization of the plugin class."""

    self.args = args
    self.kwargs = kwargs

load(self, ctx)

Called on load of the plugin from within LiquidBT.

If you are overriding this, you should set self.ctx to the passed argument.

Source code in liquidbt/plugins.py
13
14
15
16
17
18
19
20
def load(self, ctx):
    """
    Called on load of the plugin from within LiquidBT.

    If you are overriding this, **you should set
    `self.ctx` to the passed argument.**
    """
    self.ctx = ctx

shutdown(self)

Called on shutdown of LiquidBT.

At this point, no new tasks can be created, so just run any final code here.

Source code in liquidbt/plugins.py
22
23
24
25
26
27
28
29
def shutdown(self):
    """
    Called on shutdown of LiquidBT.

    At this point, no new tasks can be created,
    so just run any final code here.
    """
    pass

progressbars

CustomProgressBar

Our custom progress bar.

after_task(self)

Runs after a task.

Source code in liquidbt/progressbars.py
42
43
44
45
def after_task(self):
    """Runs after a task."""

    self.next()

before_task(self, task)

Runs before a task.

Source code in liquidbt/progressbars.py
36
37
38
39
40
def before_task(self, task):
    """Runs before a task."""

    self.max = self.ctx.task_count()
    self.do_output(task.name)

do_output(self, output)

Logs output if the shell is not interactive, or sets the title of the progress bar to the value of output if it is.

Source code in liquidbt/progressbars.py
17
18
19
20
21
22
23
24
25
26
def do_output(self, output):
    """
    Logs `output` if the shell is not interactive, or sets the
    title of the progress bar to the value of `output` if it is.
    """

    if self.interactive:
        self.message = output
    else:
        print("log " + output)

noop(self)

Does nothing.

Source code in liquidbt/progressbars.py
28
29
def noop(self):
    """Does nothing."""

post_init(self)

Reconfigure the bar if this isn't an interactive terminal.

Source code in liquidbt/progressbars.py
 8
 9
10
11
12
13
14
15
def post_init(self):
    """Reconfigure the bar if this isn't an interactive terminal."""

    self.interactive = stdout.isatty()
    if not self.interactive:
        self.update = self.noop

    return self

set_ctx(self, ctx)

Sets the build context.

Source code in liquidbt/progressbars.py
31
32
33
34
def set_ctx(self, ctx):
    """Sets the build context."""

    self.ctx = ctx

tasks

The tasks API.

RunContext

A mechanism used to manage and store data during runtime.

__init__(self, command, bar) special

Create the class. THIS IS DONE INTERNALLY, DO NOT USE!

Source code in liquidbt/tasks.py
67
68
69
70
71
72
73
74
75
def __init__(self, command: str, bar: CustomProgressBar):
    """Create the class. THIS IS DONE INTERNALLY, DO NOT USE!"""

    self._tasks = []
    self._plugins = []
    self._transformers = []
    self.command = command
    self._progress_bar = bar
    self._progress_bar.set_ctx(self)

add_task(self, task)

Adds a task to the task list.

Source code in liquidbt/tasks.py
77
78
79
80
def add_task(self, task: Task):
    """Adds a task to the task list."""

    self._tasks.append(task)

add_transformer(self, transformer)

Adds a transformer to the end of the list.

Source code in liquidbt/tasks.py
92
93
94
95
def add_transformer(self, transformer: Callable):
    """Adds a transformer to the end of the list."""

    self._transformers.append(transformer)

get_build_transformers(self)

Get a list of transformers in the order they were registered in.

Source code in liquidbt/tasks.py
 97
 98
 99
100
def get_build_transformers(self):
    """Get a list of transformers in the order they were registered in."""

    return self._transformers

get_tasks(self)

Gets the list of tasks registered.

Source code in liquidbt/tasks.py
87
88
89
90
def get_tasks(self) -> List[Task]:
    """Gets the list of tasks registered."""

    return self._tasks

task_count(self)

Get the number of tasks registered total.

Source code in liquidbt/tasks.py
82
83
84
85
def task_count(self) -> int:
    """Get the number of tasks registered total."""

    return len(self._tasks)

Task

A task that can be run.

__call__(self) special

Run the task.

Source code in liquidbt/tasks.py
47
48
49
50
51
52
def __call__(self):
    """Run the task."""

    if self.status is not TaskStatuses.READY:
        return
    self.run()

__init__(self) special

Create the class.

Source code in liquidbt/tasks.py
27
28
29
def __init__(self):
    """Create the class."""
    self.status = TaskStatuses.READY

run(self)

Run the task.

Add the code that your task should execute here, but DO NOT manually call this function anywhere in your plugin.

Source code in liquidbt/tasks.py
34
35
36
37
38
39
40
def run(self):
    """
    Run the task.

    Add the code that your task should execute here, but
    DO NOT manually call this function anywhere in your plugin.
    """

skip(self)

Set the task to skipped, preventing it from being run.

Source code in liquidbt/tasks.py
42
43
44
45
def skip(self):
    """Set the task to skipped, preventing it from being run."""

    self.status = TaskStatuses.SKIPPED

TaskStatuses

An enum for the task statuses.

COMPLETED

The task was skipped.

READY

The task has already finished.

create_task(name, runnable)

Creates a task named the value of the name argument, which calls the runnable argument.

Source code in liquidbt/tasks.py
103
104
105
106
107
108
109
110
111
112
113
114
115
def create_task(name: str, runnable: Callable) -> Task:
    """
    Creates a task named the value of the `name` argument,
    which calls the `runnable` argument.
    """

    class VirtualTask(Task):
        name: str = name

        def run(self):
            runnable()

    return VirtualTask()

Last update: