#![deny(missing_docs, dead_code)]
use std::{collections::HashMap, io::Read, path::Path, str::FromStr, sync::Arc};
use indexmap::IndexSet;
use rattler_conda_types::{Platform, RepoDataRecord};
mod builder;
mod channel;
mod conda;
mod file_format_version;
mod hash;
pub mod options;
mod parse;
mod pypi;
mod pypi_indexes;
pub mod source;
mod url_or_path;
mod utils;
pub use builder::{LockFileBuilder, LockedPackage};
pub use channel::Channel;
pub use conda::{
CondaBinaryData, CondaPackageData, CondaSourceData, ConversionError, GitShallowSpec, InputHash,
PackageBuildSource, VariantValue,
};
pub use file_format_version::FileFormatVersion;
pub use hash::PackageHashes;
pub use options::{PypiPrereleaseMode, SolveOptions};
pub use parse::ParseCondaLockError;
pub use pypi::{PypiPackageData, PypiPackageEnvironmentData, PypiSourceTreeHashable};
pub use pypi_indexes::{FindLinksUrlOrPath, PypiIndexes};
pub use rattler_conda_types::Matches;
pub use url_or_path::UrlOrPath;
pub const DEFAULT_ENVIRONMENT_NAME: &str = "default";
#[derive(Clone, Default, Debug)]
pub struct LockFile {
inner: Arc<LockFileInner>,
}
#[derive(Default, Debug)]
struct LockFileInner {
version: FileFormatVersion,
environments: Vec<EnvironmentData>,
conda_packages: Vec<CondaPackageData>,
pypi_packages: Vec<PypiPackageData>,
pypi_environment_package_data: Vec<PypiPackageEnvironmentData>,
environment_lookup: ahash::HashMap<String, usize>,
}
#[derive(Clone, Copy, Debug, Hash, Eq, PartialEq)]
enum EnvironmentPackageData {
Conda(usize),
Pypi(usize, usize),
}
#[derive(Clone, Debug)]
struct EnvironmentData {
channels: Vec<Channel>,
indexes: Option<PypiIndexes>,
options: SolveOptions,
packages: ahash::HashMap<Platform, IndexSet<EnvironmentPackageData>>,
}
impl LockFile {
pub fn builder() -> LockFileBuilder {
LockFileBuilder::new()
}
pub fn from_reader(mut reader: impl Read) -> Result<Self, ParseCondaLockError> {
let mut str = String::new();
reader.read_to_string(&mut str)?;
Self::from_str(&str)
}
pub fn from_path(path: &Path) -> Result<Self, ParseCondaLockError> {
let source = std::fs::read_to_string(path)?;
Self::from_str(&source)
}
pub fn to_path(&self, path: &Path) -> Result<(), std::io::Error> {
let file = std::fs::File::create(path)?;
serde_yaml::to_writer(file, self).map_err(std::io::Error::other)
}
pub fn render_to_string(&self) -> Result<String, std::io::Error> {
serde_yaml::to_string(self).map_err(std::io::Error::other)
}
pub fn environment(&self, name: &str) -> Option<Environment<'_>> {
let index = *self.inner.environment_lookup.get(name)?;
Some(Environment {
lock_file: self,
index,
})
}
pub fn default_environment(&self) -> Option<Environment<'_>> {
self.environment(DEFAULT_ENVIRONMENT_NAME)
}
pub fn environments(&self) -> impl ExactSizeIterator<Item = (&str, Environment<'_>)> + '_ {
self.inner
.environment_lookup
.iter()
.map(move |(name, index)| {
(
name.as_str(),
Environment {
lock_file: self,
index: *index,
},
)
})
}
pub fn version(&self) -> FileFormatVersion {
self.inner.version
}
pub fn is_empty(&self) -> bool {
self.inner.conda_packages.is_empty() && self.inner.pypi_packages.is_empty()
}
}
#[derive(Clone, Copy)]
pub struct Environment<'lock> {
lock_file: &'lock LockFile,
index: usize,
}
impl<'lock> Environment<'lock> {
fn data(&self) -> &'lock EnvironmentData {
&self.lock_file.inner.environments[self.index]
}
pub fn lock_file(&self) -> &'lock LockFile {
self.lock_file
}
pub fn platforms(&self) -> impl ExactSizeIterator<Item = Platform> + '_ {
self.data().packages.keys().copied()
}
pub fn channels(&self) -> &[Channel] {
&self.data().channels
}
pub fn pypi_indexes(&self) -> Option<&PypiIndexes> {
self.data().indexes.as_ref()
}
pub fn pypi_prerelease_mode(&self) -> Option<PypiPrereleaseMode> {
self.data().options.pypi_prerelease_mode
}
pub fn solve_options(&self) -> &SolveOptions {
&self.data().options
}
pub fn packages(
&self,
platform: Platform,
) -> Option<impl DoubleEndedIterator<Item = LockedPackageRef<'lock>> + ExactSizeIterator + '_>
{
Some(
self.data()
.packages
.get(&platform)?
.iter()
.map(move |package| match package {
EnvironmentPackageData::Conda(data) => {
LockedPackageRef::Conda(&self.lock_file.inner.conda_packages[*data])
}
EnvironmentPackageData::Pypi(data, env_data) => LockedPackageRef::Pypi(
&self.lock_file.inner.pypi_packages[*data],
&self.lock_file.inner.pypi_environment_package_data[*env_data],
),
}),
)
}
pub fn packages_by_platform(
&self,
) -> impl ExactSizeIterator<
Item = (
Platform,
impl DoubleEndedIterator<Item = LockedPackageRef<'lock>> + ExactSizeIterator + '_,
),
> + '_ {
let env_data = self.data();
env_data.packages.iter().map(move |(platform, packages)| {
(
*platform,
packages.iter().map(move |package| match package {
EnvironmentPackageData::Conda(data) => {
LockedPackageRef::Conda(&self.lock_file.inner.conda_packages[*data])
}
EnvironmentPackageData::Pypi(data, env_data) => LockedPackageRef::Pypi(
&self.lock_file.inner.pypi_packages[*data],
&self.lock_file.inner.pypi_environment_package_data[*env_data],
),
}),
)
})
}
pub fn pypi_packages_by_platform(
&self,
) -> impl ExactSizeIterator<
Item = (
Platform,
impl DoubleEndedIterator<Item = (&'lock PypiPackageData, &'lock PypiPackageEnvironmentData)>,
),
> + '_ {
let env_data = self.data();
env_data.packages.iter().map(|(platform, packages)| {
let records = packages.iter().filter_map(|package| match package {
EnvironmentPackageData::Conda(_) => None,
EnvironmentPackageData::Pypi(pkg_data_idx, env_data_idx) => Some((
&self.lock_file.inner.pypi_packages[*pkg_data_idx],
&self.lock_file.inner.pypi_environment_package_data[*env_data_idx],
)),
});
(*platform, records)
})
}
pub fn conda_packages_by_platform(
&self,
) -> impl ExactSizeIterator<
Item = (
Platform,
impl DoubleEndedIterator<Item = &'lock CondaPackageData> + '_,
),
> + '_ {
self.packages_by_platform()
.map(|(platform, packages)| (platform, packages.filter_map(LockedPackageRef::as_conda)))
}
pub fn conda_repodata_records_by_platform(
&self,
) -> Result<HashMap<Platform, Vec<RepoDataRecord>>, ConversionError> {
self.conda_packages_by_platform()
.map(|(platform, packages)| {
Ok((
platform,
packages
.filter_map(CondaPackageData::as_binary)
.map(RepoDataRecord::try_from)
.collect::<Result<Vec<_>, ConversionError>>()?,
))
})
.collect()
}
pub fn conda_packages(
&self,
platform: Platform,
) -> Option<impl DoubleEndedIterator<Item = &'lock CondaPackageData> + '_> {
self.packages(platform)
.map(|packages| packages.filter_map(LockedPackageRef::as_conda))
}
pub fn conda_repodata_records(
&self,
platform: Platform,
) -> Result<Option<Vec<RepoDataRecord>>, ConversionError> {
self.conda_packages(platform)
.map(|packages| {
packages
.filter_map(CondaPackageData::as_binary)
.map(RepoDataRecord::try_from)
.collect()
})
.transpose()
}
pub fn pypi_packages(
&self,
platform: Platform,
) -> Option<
impl DoubleEndedIterator<Item = (&'lock PypiPackageData, &'lock PypiPackageEnvironmentData)>
+ '_,
> {
self.packages(platform)
.map(|pkgs| pkgs.filter_map(LockedPackageRef::as_pypi))
}
pub fn has_pypi_packages(&self, platform: Platform) -> bool {
self.pypi_packages(platform)
.is_some_and(|mut packages| packages.next().is_some())
}
pub fn to_owned(self) -> OwnedEnvironment {
OwnedEnvironment {
lock_file: self.lock_file.clone(),
index: self.index,
}
}
}
#[derive(Clone)]
pub struct OwnedEnvironment {
lock_file: LockFile,
index: usize,
}
impl OwnedEnvironment {
pub fn as_ref(&self) -> Environment<'_> {
Environment {
lock_file: &self.lock_file,
index: self.index,
}
}
pub fn lock_file(&self) -> LockFile {
self.lock_file.clone()
}
}
#[derive(Clone, Copy)]
pub enum LockedPackageRef<'lock> {
Conda(&'lock CondaPackageData),
Pypi(&'lock PypiPackageData, &'lock PypiPackageEnvironmentData),
}
impl<'lock> LockedPackageRef<'lock> {
pub fn name(self) -> &'lock str {
match self {
LockedPackageRef::Conda(data) => data.record().name.as_source(),
LockedPackageRef::Pypi(data, _) => data.name.as_ref(),
}
}
pub fn location(self) -> &'lock UrlOrPath {
match self {
LockedPackageRef::Conda(data) => data.location(),
LockedPackageRef::Pypi(data, _) => &data.location,
}
}
pub fn as_pypi(self) -> Option<(&'lock PypiPackageData, &'lock PypiPackageEnvironmentData)> {
match self {
LockedPackageRef::Conda(_) => None,
LockedPackageRef::Pypi(data, env) => Some((data, env)),
}
}
pub fn as_conda(self) -> Option<&'lock CondaPackageData> {
match self {
LockedPackageRef::Conda(data) => Some(data),
LockedPackageRef::Pypi(..) => None,
}
}
pub fn as_binary_conda(self) -> Option<&'lock CondaBinaryData> {
self.as_conda().and_then(CondaPackageData::as_binary)
}
pub fn as_source_conda(self) -> Option<&'lock CondaSourceData> {
self.as_conda().and_then(CondaPackageData::as_source)
}
}
#[cfg(test)]
mod test {
use std::{
path::{Path, PathBuf},
str::FromStr,
};
use rattler_conda_types::{Platform, RepoDataRecord};
use rstest::*;
use super::{LockFile, DEFAULT_ENVIRONMENT_NAME};
#[rstest]
#[case::v0_numpy("v0/numpy-conda-lock.yml")]
#[case::v0_python("v0/python-conda-lock.yml")]
#[case::v0_pypi_matplotlib("v0/pypi-matplotlib-conda-lock.yml")]
#[case::v3_robostack("v3/robostack-turtlesim-conda-lock.yml")]
#[case::v3_numpy("v4/numpy-lock.yml")]
#[case::v4_python("v4/python-lock.yml")]
#[case::v4_pypi_matplotlib("v4/pypi-matplotlib-lock.yml")]
#[case::v4_turtlesim("v4/turtlesim-lock.yml")]
#[case::v4_pypi_path("v4/path-based-lock.yml")]
#[case::v4_pypi_absolute_path("v4/absolute-path-lock.yml")]
#[case::v5_pypi_flat_index("v5/flat-index-lock.yml")]
#[case::v5_with_and_without_purl("v5/similar-with-and-without-purl.yml")]
#[case::v6_conda_source_path("v6/conda-path-lock.yml")]
#[case::v6_derived_channel("v6/derived-channel-lock.yml")]
#[case::v6_sources("v6/sources-lock.yml")]
#[case::v6_options("v6/options-lock.yml")]
#[case::v6_pixi_build_pinned_source("v6/pixi-build-pinned-source-lock.yml")]
#[case::v6_pixi_build_url_source("v6/pixi-build-url-source-lock.yml")]
#[case::v6_pixi_build_git_tag_source("v6/pixi-build-git-tag-source-lock.yml")]
#[case::v6_pixi_build_git_rev_only_source("v6/pixi-build-git-rev-only-source-lock.yml")]
#[case::v6_source_package_with_variants("v6/source-package-with-variants-lock.yml")]
#[case::v6_multiple_source_packages_with_variants(
"v6/multiple-source-packages-with-variants-lock.yml"
)]
fn test_parse(#[case] file_name: &str) {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock")
.join(file_name);
let conda_lock = LockFile::from_path(&path).unwrap();
insta::assert_yaml_snapshot!(file_name, conda_lock);
}
#[rstest]
fn test_roundtrip(
#[files("../../test-data/conda-lock/**/*.yml")]
#[exclude("forward-compatible-lock")]
path: PathBuf,
) {
let conda_lock = LockFile::from_path(&path).unwrap();
let rendered_lock_file = conda_lock.render_to_string().unwrap();
let parsed_lock_file = LockFile::from_str(&rendered_lock_file).unwrap();
let rerendered_lock_file = parsed_lock_file.render_to_string().unwrap();
similar_asserts::assert_eq!(rendered_lock_file, rerendered_lock_file);
}
#[test]
fn test_issue_615() {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock/v4/absolute-path-lock.yml");
let conda_lock = LockFile::from_path(&path);
assert!(conda_lock.is_ok());
}
#[test]
fn packages_for_platform() {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock")
.join("v0/numpy-conda-lock.yml");
let conda_lock = LockFile::from_path(&path).unwrap();
insta::assert_yaml_snapshot!(conda_lock
.environment(DEFAULT_ENVIRONMENT_NAME)
.unwrap()
.packages(Platform::Linux64)
.unwrap()
.map(|p| p.location().to_string())
.collect::<Vec<_>>());
insta::assert_yaml_snapshot!(conda_lock
.environment(DEFAULT_ENVIRONMENT_NAME)
.unwrap()
.packages(Platform::Osx64)
.unwrap()
.map(|p| p.location().to_string())
.collect::<Vec<_>>());
}
#[test]
fn test_has_pypi_packages() {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock")
.join("v4/pypi-matplotlib-lock.yml");
let conda_lock = LockFile::from_path(&path).unwrap();
assert!(conda_lock
.environment(DEFAULT_ENVIRONMENT_NAME)
.unwrap()
.has_pypi_packages(Platform::Linux64));
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock")
.join("v6/numpy-as-pypi-lock.yml");
let conda_lock = LockFile::from_path(&path).unwrap();
assert!(conda_lock
.environment(DEFAULT_ENVIRONMENT_NAME)
.unwrap()
.has_pypi_packages(Platform::OsxArm64));
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock")
.join("v6/python-from-conda-only-lock.yml");
let conda_lock = LockFile::from_path(&path).unwrap();
assert!(!conda_lock
.environment(DEFAULT_ENVIRONMENT_NAME)
.unwrap()
.has_pypi_packages(Platform::OsxArm64));
}
#[test]
fn test_is_empty() {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock")
.join("v6/empty-lock.yml");
let conda_lock = LockFile::from_path(&path).unwrap();
assert!(conda_lock.is_empty());
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/conda-lock")
.join("v6/python-from-conda-only-lock.yml");
let conda_lock = LockFile::from_path(&path).unwrap();
assert!(!conda_lock.is_empty());
}
#[test]
fn solve_roundtrip() {
let path = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("../../test-data/repodata-records/_libgcc_mutex-0.1-conda_forge.json");
let content = std::fs::read_to_string(&path).unwrap();
let repodata_record: RepoDataRecord = serde_json::from_str(&content).unwrap();
assert_eq!(repodata_record.package_record.arch, None);
assert_eq!(repodata_record.package_record.platform, None);
let lock_file = LockFile::builder()
.with_conda_package(
DEFAULT_ENVIRONMENT_NAME,
Platform::Linux64,
repodata_record.clone().into(),
)
.finish();
let rendered_lock_file = lock_file.render_to_string().unwrap();
let parsed_lock_file = LockFile::from_str(&rendered_lock_file).unwrap();
let repodata_records = parsed_lock_file
.environment(DEFAULT_ENVIRONMENT_NAME)
.unwrap()
.conda_repodata_records(Platform::Linux64)
.unwrap()
.unwrap();
let repodata_record_two = repodata_records[0].clone();
assert_eq!(
repodata_record_two.package_record.arch,
Some("x86_64".to_string())
);
assert_eq!(
repodata_record_two.package_record.platform,
Some("linux".to_string())
);
let rerendered_lock_file_two = parsed_lock_file.render_to_string().unwrap();
assert_eq!(rendered_lock_file, rerendered_lock_file_two);
}
}