Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 50 additions & 1 deletion component_config/configSchema.json
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,55 @@
}
}
},
"private_pypi": {
"type": "object",
"title": "Private PyPI Repository (Optional)",
"propertyOrder": 55,
"options": {
"tooltip": "Configure a private PyPI repository (e.g., Artifactory, Nexus, or any PEP 503 compliant index) to install packages from. The private repository will be used as an additional index alongside the public PyPI."
},
"properties": {
"enabled": {
"type": "boolean",
"title": "Enable Private PyPI Repository",
"default": false,
"propertyOrder": 1
},
"url": {
"type": "string",
"title": "Repository URL",
"propertyOrder": 2,
"options": {
"tooltip": "The URL of the private PyPI repository (e.g., https://artifactory.example.com/api/pypi/pypi-local/simple/). Must be a PEP 503 compliant simple index URL.",
"dependencies": {
"enabled": true
}
}
},
"username": {
"type": "string",
"title": "Username",
"propertyOrder": 3,
"options": {
"tooltip": "Username for authentication (optional for some repositories).",
"dependencies": {
"enabled": true
}
}
},
"#password": {
"type": "string",
"title": "Password / Token",
"propertyOrder": 4,
"options": {
"tooltip": "Password or API token for authentication.",
"dependencies": {
"enabled": true
}
}
}
}
},
"git": {
"type": "object",
"title": "Git Repository Source Settings",
Expand Down Expand Up @@ -195,4 +244,4 @@
}
}
}
}
}
4 changes: 2 additions & 2 deletions src/component.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,9 +76,9 @@ def run(self):
if self.parameters.source == SourceEnum.CODE:
if "keboola.component" not in self.parameters.packages:
self.parameters.packages.insert(0, "keboola.component")
PackageInstaller.install_packages(self.parameters.packages)
PackageInstaller.install_packages(self.parameters.packages, self.parameters.private_pypi)
else:
PackageInstaller.install_packages_for_repository(base_path)
PackageInstaller.install_packages_for_repository(base_path, self.parameters.private_pypi)

self._merge_user_parameters()

Expand Down
9 changes: 9 additions & 0 deletions src/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,14 @@ class GitConfiguration:
ssh_keys: SSHKeysConfiguration = field(default_factory=SSHKeysConfiguration)


@dataclass
class PrivatePyPIConfiguration:
enabled: bool = False
url: str = ""
username: str = ""
encrypted_password: str | None = None


@dataclass
class Configuration:
source: SourceEnum = SourceEnum.CODE
Expand All @@ -57,6 +65,7 @@ class Configuration:
packages: list[str] = field(default_factory=list)
code: str = ""
git: GitConfiguration = field(default_factory=GitConfiguration)
private_pypi: PrivatePyPIConfiguration = field(default_factory=PrivatePyPIConfiguration)

def __post_init__(self):
if isinstance(self.user_properties, list):
Expand Down
51 changes: 45 additions & 6 deletions src/package_installer.py
Original file line number Diff line number Diff line change
@@ -1,42 +1,81 @@
import logging
import os
from pathlib import Path
from typing import TYPE_CHECKING

from subprocess_runner import SubprocessRunner

if TYPE_CHECKING:
from configuration import PrivatePyPIConfiguration

MSG_OK = "Installation successful."
MSG_ERR = "Installation failed."

PRIVATE_PYPI_INDEX_NAME = "private"


class PackageInstaller:
@staticmethod
def install_packages(packages: list[str]):
def _setup_private_pypi_env(private_pypi: "PrivatePyPIConfiguration | None") -> None:
"""
Set up environment variables for private PyPI repository authentication.

Args:
private_pypi: Private PyPI configuration, or None if not configured.
"""
if private_pypi is None or not private_pypi.enabled or not private_pypi.url:
return

logging.info("Configuring private PyPI repository: %s", private_pypi.url)

os.environ["UV_INDEX"] = f"{PRIVATE_PYPI_INDEX_NAME}={private_pypi.url}"

if private_pypi.username:
env_var_name = f"UV_INDEX_{PRIVATE_PYPI_INDEX_NAME.upper()}_USERNAME"
os.environ[env_var_name] = private_pypi.username

if private_pypi.encrypted_password:
env_var_name = f"UV_INDEX_{PRIVATE_PYPI_INDEX_NAME.upper()}_PASSWORD"
os.environ[env_var_name] = private_pypi.encrypted_password

@staticmethod
def install_packages(
packages: list[str],
private_pypi: "PrivatePyPIConfiguration | None" = None
) -> None:
PackageInstaller._setup_private_pypi_env(private_pypi)

for package in packages:
logging.info("Installing package: %s...", package)
args = ["uv", "pip", "install", package]
SubprocessRunner.run(args, MSG_OK, MSG_ERR)

@staticmethod
def install_packages_for_repository(repository_path: Path):
def install_packages_for_repository(
repository_path: Path,
private_pypi: "PrivatePyPIConfiguration | None" = None
) -> None:
"""
Install packages based on the given repository path.
- If there is a pyproject.toml and a uv.lock file, run uv sync.
- If there is a requirements.txt file, install packages from it using uv.

Args:
repository_path (str): Path to the repository containing requirements.txt.
repository_path: Path to the repository containing requirements.txt.
private_pypi: Private PyPI configuration, or None if not configured.
"""
PackageInstaller._setup_private_pypi_env(private_pypi)

pyproject_file = repository_path / "pyproject.toml"
uv_lock_file = repository_path / "uv.lock"
requirements_file = repository_path / "requirements.txt"

# Explicitly install keboola.component in case user didn't include in their dependencies file
PackageInstaller.install_packages(["keboola.component"])
PackageInstaller.install_packages(["keboola.component"], private_pypi)

args = None
if pyproject_file.exists() and uv_lock_file.exists():
logging.info("Running uv sync...")
os.chdir(repository_path) # it is currently impossible to pass custom uv.lock path
os.chdir(repository_path)
args = ["uv", "sync", "--inexact"]
elif requirements_file.exists():
logging.info("Installing packages from requirements.txt...")
Expand Down