def reactivate_local_pixi(
*,
task_group_activity_id: int,
task_group_id: int,
resource: Resource,
profile: Profile,
) -> None:
"""
Reactivate a task group venv.
This function is run as a background task, therefore exceptions must be
handled.
Args:
task_group_id:
task_group_activity_id:
resource:
"""
LOGGER_NAME = f"{__name__}.ID{task_group_activity_id}"
with TemporaryDirectory() as tmpdir:
log_file_path = get_log_path(Path(tmpdir))
logger = set_logger(
logger_name=LOGGER_NAME,
log_file_path=log_file_path,
)
logger.debug("START")
with next(get_sync_db()) as db:
try:
task_group, activity = get_activity_and_task_group(
task_group_activity_id=task_group_activity_id,
task_group_id=task_group_id,
db=db,
logger_name=LOGGER_NAME,
)
except NoResultFound:
return
source_dir = Path(task_group.path, SOURCE_DIR_NAME).as_posix()
if Path(source_dir).exists():
error_msg = f"{source_dir} already exists."
logger.error(error_msg)
fail_and_cleanup(
task_group=task_group,
task_group_activity=activity,
logger_name=LOGGER_NAME,
log_file_path=log_file_path,
exception=FileExistsError(error_msg),
db=db,
)
return
try:
activity.status = TaskGroupActivityStatus.ONGOING
activity = add_commit_refresh(obj=activity, db=db)
pixi_home = resource.tasks_pixi_config["versions"][
task_group.pixi_version
]
pixi_cache_dir = profile.pixi_cache_dir or os.path.join(
pixi_home, "cache"
)
logger.info(f"Setting PIXI_CACHE_DIR to {pixi_cache_dir}")
common_args = dict(
replacements={
("__PIXI_HOME__", pixi_home),
("__PIXI_CACHE_DIR__", pixi_cache_dir),
("__PACKAGE_DIR__", task_group.path),
("__TAR_GZ_PATH__", task_group.archive_path),
(
"__IMPORT_PACKAGE_NAME__",
task_group.pkg_name.replace("-", "_"),
),
("__SOURCE_DIR_NAME__", SOURCE_DIR_NAME),
("__FROZEN_OPTION__", "--frozen"),
(
"__TOKIO_WORKER_THREADS__",
str(
resource.tasks_pixi_config[
"TOKIO_WORKER_THREADS"
]
),
),
(
"__PIXI_CONCURRENT_SOLVES__",
str(
resource.tasks_pixi_config[
"PIXI_CONCURRENT_SOLVES"
]
),
),
(
"__PIXI_CONCURRENT_DOWNLOADS__",
str(
resource.tasks_pixi_config[
"PIXI_CONCURRENT_DOWNLOADS"
]
),
),
},
script_dir=Path(
task_group.path, SCRIPTS_SUBFOLDER
).as_posix(),
prefix=(
f"{int(time.time())}_"
f"{TaskGroupActivityAction.REACTIVATE}"
),
logger_name=LOGGER_NAME,
)
# Run script 1 - extract tar.gz into `source_dir`
_customize_and_run_template(
template_filename="pixi_1_extract.sh",
**common_args,
)
activity.log = get_current_log(log_file_path)
activity = add_commit_refresh(obj=activity, db=db)
# Simplify `pyproject.toml`
pyproject_toml_path = Path(source_dir, "pyproject.toml")
edit_pyproject_toml_in_place_local(
pyproject_toml_path, resource=resource
)
# Write pixi.lock into `source_dir`
logger.debug(f"start - writing {source_dir}/pixi.lock")
with Path(source_dir, "pixi.lock").open("w") as f:
f.write(task_group.env_info)
logger.debug(f"end - writing {source_dir}/pixi.lock")
# Run script 2 - run pixi-install command
_customize_and_run_template(
template_filename="pixi_2_install.sh",
**common_args,
)
activity.log = get_current_log(log_file_path)
activity = add_commit_refresh(obj=activity, db=db)
# Run script 3 - post-install
_customize_and_run_template(
template_filename="pixi_3_post_install.sh",
**common_args,
)
activity.log = get_current_log(log_file_path)
activity = add_commit_refresh(obj=activity, db=db)
# Make task folder 755
source_dir = Path(task_group.path, SOURCE_DIR_NAME).as_posix()
command = f"chmod -R 755 {source_dir}"
execute_command_sync(
command=command,
logger_name=LOGGER_NAME,
)
activity.log = get_current_log(log_file_path)
activity.status = TaskGroupActivityStatus.OK
activity.timestamp_ended = get_timestamp()
activity = add_commit_refresh(obj=activity, db=db)
task_group.active = True
task_group = add_commit_refresh(obj=task_group, db=db)
logger.debug("END")
reset_logger_handlers(logger)
except Exception as reactivate_e:
rmtree_nofail(
folder_path=source_dir,
logger_name=LOGGER_NAME,
)
fail_and_cleanup(
task_group=task_group,
task_group_activity=activity,
logger_name=LOGGER_NAME,
log_file_path=log_file_path,
exception=reactivate_e,
db=db,
)
return