use std::fmt::Write;
use std::path::Path;
use anyhow::{Result, bail};
use owo_colors::OwoColorize;
use tracing::debug;
use uv_python::downloads::ManagedPythonDownloadList;
use uv_cache::Cache;
use uv_client::BaseClientBuilder;
use uv_configuration::DependencyGroupsWithDefaults;
use uv_fs::Simplified;
use uv_preview::Preview;
use uv_python::{
EnvironmentPreference, PYTHON_VERSION_FILENAME, PythonDownloads, PythonInstallation,
PythonPreference, PythonRequest, PythonVersionFile, VersionFileDiscoveryOptions,
};
use uv_settings::PythonInstallMirrors;
use uv_warnings::warn_user_once;
use uv_workspace::{DiscoveryOptions, VirtualProject, WorkspaceCache};
use crate::commands::{
ExitStatus, project::find_requires_python, reporters::PythonDownloadReporter,
};
use crate::printer::Printer;
#[expect(clippy::fn_params_excessive_bools)]
pub(crate) async fn pin(
project_dir: &Path,
request: Option<String>,
resolved: bool,
python_preference: PythonPreference,
python_downloads: PythonDownloads,
no_project: bool,
global: bool,
rm: bool,
install_mirrors: PythonInstallMirrors,
client_builder: BaseClientBuilder<'_>,
cache: &Cache,
workspace_cache: &WorkspaceCache,
printer: Printer,
preview: Preview,
) -> Result<ExitStatus> {
let virtual_project = if no_project {
None
} else {
match VirtualProject::discover(project_dir, &DiscoveryOptions::default(), workspace_cache)
.await
{
Ok(virtual_project) => Some(virtual_project),
Err(err) => {
debug!("Failed to discover virtual project: {err}");
None
}
}
};
let version_file = PythonVersionFile::discover(
project_dir,
&VersionFileDiscoveryOptions::default().with_no_local(global),
)
.await;
if rm {
let Some(file) = version_file? else {
if global {
bail!("No global Python pin found");
}
bail!("No Python version file found");
};
if !global && file.is_global() {
bail!("No Python version file found; use `--rm --global` to remove the global pin");
}
fs_err::tokio::remove_file(file.path()).await?;
writeln!(
printer.stdout(),
"Removed {} at `{}`",
if global {
"global Python pin"
} else {
"Python version file"
},
file.path().user_display()
)?;
return Ok(ExitStatus::Success);
}
let Some(request) = request else {
if let Some(file) = version_file? {
for pin in file.versions() {
writeln!(printer.stdout(), "{}", pin.to_canonical_string())?;
if let Some(virtual_project) = &virtual_project {
let client = client_builder.clone().retries(0).build()?;
let download_list = ManagedPythonDownloadList::new(
&client,
install_mirrors.python_downloads_json_url.as_deref(),
)
.await?;
warn_if_existing_pin_incompatible_with_project(
pin,
virtual_project,
python_preference,
&download_list,
cache,
preview,
);
}
}
return Ok(ExitStatus::Success);
}
bail!("No Python version file found; specify a version to create one")
};
let request = PythonRequest::parse(&request);
if let PythonRequest::ExecutableName(name) = request {
bail!("Requests for arbitrary names (e.g., `{name}`) are not supported in version files");
}
let reporter = PythonDownloadReporter::single(printer);
let python = match PythonInstallation::find_or_download(
Some(&request),
EnvironmentPreference::OnlySystem,
python_preference,
python_downloads,
&client_builder,
cache,
Some(&reporter),
install_mirrors.python_install_mirror.as_deref(),
install_mirrors.pypy_install_mirror.as_deref(),
install_mirrors.python_downloads_json_url.as_deref(),
preview,
)
.await
{
Ok(python) => Some(python),
Err(uv_python::Error::MissingPython(err, ..)) if !resolved => {
warn_user_once!("{err}");
None
}
Err(err) if !resolved => {
debug!("{err}");
None
}
Err(err) => return Err(err.into()),
};
if let Some(virtual_project) = &virtual_project {
if let Some(request_version) = request.as_pep440_version() {
assert_pin_compatible_with_project(
&Pin {
request: &request,
version: &request_version,
resolved: false,
existing: false,
},
virtual_project,
)?;
} else {
if let Some(python) = &python {
if let Err(err) = assert_pin_compatible_with_project(
&Pin {
request: &request,
version: python.python_version(),
resolved: true,
existing: false,
},
virtual_project,
) {
if resolved {
return Err(err);
}
warn_user_once!("{err}");
}
}
}
}
let request = if resolved {
PythonRequest::parse(
&python
.unwrap()
.interpreter()
.sys_executable()
.user_display()
.to_string(),
)
} else {
request
};
let existing = version_file.ok().flatten();
let new = if global {
let Some(new) = PythonVersionFile::global() else {
bail!("Failed to determine directory for global Python pin");
};
new.with_versions(vec![request])
} else {
PythonVersionFile::new(project_dir.join(PYTHON_VERSION_FILENAME))
.with_versions(vec![request])
};
new.write().await?;
if let Some(existing) = existing
.as_ref()
.filter(|existing| existing.path() == new.path())
.and_then(PythonVersionFile::version)
.filter(|version| *version != new.version().unwrap())
{
writeln!(
printer.stdout(),
"Updated `{}` from `{}` -> `{}`",
new.path().user_display().cyan(),
existing.to_canonical_string().green(),
new.version().unwrap().to_canonical_string().green()
)?;
} else {
writeln!(
printer.stdout(),
"Pinned `{}` to `{}`",
new.path().user_display().cyan(),
new.version().unwrap().to_canonical_string().green()
)?;
}
Ok(ExitStatus::Success)
}
fn warn_if_existing_pin_incompatible_with_project(
pin: &PythonRequest,
virtual_project: &VirtualProject,
python_preference: PythonPreference,
downloads_list: &ManagedPythonDownloadList,
cache: &Cache,
preview: Preview,
) {
if let Some(pin_version) = pin.as_pep440_version() {
if let Err(err) = assert_pin_compatible_with_project(
&Pin {
request: pin,
version: &pin_version,
resolved: false,
existing: true,
},
virtual_project,
) {
warn_user_once!("{err}");
return;
}
}
match PythonInstallation::find(
pin,
EnvironmentPreference::OnlySystem,
python_preference,
downloads_list,
cache,
preview,
) {
Ok(python) => {
let python_version = python.python_version();
debug!(
"The pinned Python version `{}` resolves to `{}`",
pin, python_version
);
if let Err(err) = assert_pin_compatible_with_project(
&Pin {
request: pin,
version: python_version,
resolved: true,
existing: true,
},
virtual_project,
) {
warn_user_once!("{err}");
}
}
Err(err) => {
warn_user_once!(
"Failed to resolve pinned Python version `{}`: {err}",
pin.to_canonical_string(),
);
}
}
}
struct Pin<'a> {
request: &'a PythonRequest,
version: &'a uv_pep440::Version,
resolved: bool,
existing: bool,
}
fn assert_pin_compatible_with_project(pin: &Pin, virtual_project: &VirtualProject) -> Result<()> {
let groups = DependencyGroupsWithDefaults::none();
let (requires_python, project_type) = match virtual_project {
VirtualProject::Project(project_workspace) => {
debug!(
"Discovered project `{}` at: {}",
project_workspace.project_name(),
project_workspace.workspace().install_path().display()
);
let requires_python = find_requires_python(project_workspace.workspace(), &groups)?;
(requires_python, "project")
}
VirtualProject::NonProject(workspace) => {
debug!(
"Discovered virtual workspace at: {}",
workspace.install_path().display()
);
let requires_python = find_requires_python(workspace, &groups)?;
(requires_python, "workspace")
}
};
let Some(requires_python) = requires_python else {
return Ok(());
};
if requires_python.contains(pin.version) {
return Ok(());
}
let given = if pin.existing { "pinned" } else { "requested" };
let resolved = if pin.resolved {
format!(" resolves to `{}` which ", pin.version)
} else {
String::new()
};
Err(anyhow::anyhow!(
"The {given} Python version `{}`{resolved} is incompatible with the {} `requires-python` value of `{}`.",
pin.request.to_canonical_string(),
project_type,
requires_python
))
}