tasks.py (55 lines of code) (raw):
# Copyright 2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import glob
import os
import components.processing.tasks
import components.webui.tasks
from invoke import Collection, task
from uv import find_uv_bin
# Find the base directory for invoke
BASE_DIR = os.path.dirname(__file__)
########
#
# Maintaining the virtual environment tasks
#
VENV_DIR = os.path.join(BASE_DIR, ".venv")
REQS_ALL = os.path.join(BASE_DIR, "reqs", "requirements_all.txt")
@task
def sync(c, upgrade=False, upgrade_package=""):
"""Synchronise environments."""
c.run(f'"{find_uv_bin()}" pip sync -q "{REQS_ALL}"')
@task(
help={
"upgrade": "Upgrade all dependencies (subject to contraints)",
"upgrade-package": "Upgrade the specific dependency",
"quiet": "Quiet output",
},
post=[sync],
)
def lock(c, upgrade=False, upgrade_package="", quiet=True):
"""Lock and synchronise environments."""
REQ_ALL_IN = os.path.join(BASE_DIR, "reqs", "requirements_all.in")
opt_quiet = "-q" if quiet else ""
opt_upgrade = ""
opt_upgrade_package = ""
if upgrade:
opt_upgrade = "--upgrade "
if upgrade_package:
opt_upgrade_package = f"--upgrade-package {upgrade_package}"
# Lock the consolidated requirements
c.run(
f'"{find_uv_bin()}" pip compile '
f"{opt_quiet} {opt_upgrade} {opt_upgrade_package} "
f'"{os.path.relpath(REQ_ALL_IN, os.getcwd())}" -o "{os.path.relpath(REQS_ALL, os.getcwd())}"'
)
# Generate a constraints file
CONSTRAINTS = os.path.join(BASE_DIR, "reqs", "constraints.txt")
c.run(f'cat "{REQS_ALL}" | grep -v "^-e " > "{CONSTRAINTS}"')
# Gather all requirements
reqs_in = list(
glob.glob(os.path.join(BASE_DIR, "**", "pyproject.toml"), recursive=True)
)
reqs_in += list(
glob.glob(os.path.join(BASE_DIR, "**", "requirements.in"), recursive=True)
)
# Re-generate the requirements.txt for specific deployments
# (honouring consolidated requirements)
REL_CONSTRAINTS = os.path.relpath(CONSTRAINTS, os.getcwd())
for req_in in reqs_in:
req_txt = os.path.join(os.path.dirname(req_in), "requirements.txt")
if os.path.exists(req_txt):
c.run(
f'"{find_uv_bin()}" pip compile {opt_quiet} --generate-hashes '
f'-c "{REL_CONSTRAINTS}" "{os.path.relpath(req_in, os.getcwd())}" | '
'grep -v "\\${PROJECT_ROOT}" >'
f'"{req_txt}"'
)
#
# Build collection from all available tasks
#
# Root tasks
namespace = Collection(sync, lock)
# Sub-tasks
namespace.add_collection(components.processing.tasks, name="processing")
namespace.add_collection(components.webui.tasks, name="webui")