Files
ntsk 9d1070091a fix(manager/pep621): support array form of tool.uv.sources (#44970)
* test(manager/pep621): cover array uv sources

* fix(manager/pep621): support array uv sources

* test(manager/pep621): cover unknown source fallback
2026-09-04 11:06:07 +00:00

414 lines
12 KiB
TypeScript

import { isString } from '@sindresorhus/is';
import { quote } from 'shlex';
import { TEMPORARY_ERROR } from '../../../../constants/error-messages.ts';
import { logger } from '../../../../logger/index.ts';
import type { HostRule } from '../../../../types/index.ts';
import { coerceArray } from '../../../../util/array.ts';
import type {
ExecOptions,
ToolConstraint,
} from '../../../../util/exec/types.ts';
import {
findLocalSiblingOrParent,
readLocalFile,
} from '../../../../util/fs/index.ts';
import { withGitEnvironment } from '../../../../util/git/exec.ts';
import { find } from '../../../../util/host-rules.ts';
import { regEx } from '../../../../util/regex.ts';
import { Result } from '../../../../util/result.ts';
import { parseUrl } from '../../../../util/url.ts';
import { PypiDatasource } from '../../../datasource/pypi/index.ts';
import { getGoogleAuthHostRule } from '../../../datasource/util.ts';
import type {
PackageDependency,
UpdateArtifact,
UpdateArtifactsResult,
Upgrade,
} from '../../types.ts';
import { applyGitSource } from '../../util.ts';
import { type PyProject, UvLockfile, type UvSource } from '../schema.ts';
import { depTypes } from '../utils.ts';
import { BasePyProjectProcessor } from './abstract.ts';
const uvUpdateCMD = 'uv lock';
const gitExec = withGitEnvironment(['pep621']);
function isUvIndexSource(
source: UvSource,
): source is Extract<UvSource, { index: string }> {
return 'index' in source;
}
export class UvProcessor extends BasePyProjectProcessor {
override lockfileName = 'uv.lock';
process(project: PyProject, deps: PackageDependency[]): PackageDependency[] {
const uv = project.tool?.uv;
if (!uv) {
return deps;
}
const hasExplicitDefault = uv.index?.some(
(index) => index.default && index.explicit,
);
const defaultIndex = uv.index?.find(
(index) => index.default && !index.explicit,
);
const implicitIndexUrls = uv.index
?.filter((index) => !index.explicit && index.name !== defaultIndex?.name)
?.map(({ url }) => url);
const devDependencies = uv['dev-dependencies'];
if (devDependencies) {
deps.push(...devDependencies);
}
// https://docs.astral.sh/uv/concepts/dependencies/#dependency-sources
// Skip sources that do not make sense to handle (e.g. path).
if (uv.sources || defaultIndex || implicitIndexUrls) {
for (const dep of deps) {
/* v8 ignore next -- needs test */
if (!dep.packageName) {
continue;
}
if (dep.depType === 'requires-python') {
continue;
}
// Using `packageName` as it applies PEP 508 normalization, which is
// also applied by uv when matching a source to a dependency.
const depSources = uv.sources?.[dep.packageName];
if (depSources) {
// Dependency is pinned to one or more specific sources.
dep.depType = depTypes.uvSources;
if (depSources.every(isUvIndexSource)) {
// Sources referencing an index, possibly disambiguated by
// environment markers. Any of the indexes can serve the package,
// so use all of them as registries.
const registryUrls: string[] = [];
for (const depSource of depSources) {
const index = uv.index?.find(
({ name }) => name === depSource.index,
);
if (index) {
registryUrls.push(index.url);
}
}
if (registryUrls.length) {
dep.registryUrls = [...new Set(registryUrls)];
}
} else if (depSources.length === 1) {
const depSource = depSources[0];
if ('git' in depSource) {
applyGitSource(
dep,
depSource.git,
depSource.rev,
depSource.tag,
depSource.branch,
);
} else if ('url' in depSource) {
dep.skipReason = 'unsupported-url';
} else if ('path' in depSource) {
dep.skipReason = 'path-dependency';
} else if ('workspace' in depSource) {
dep.skipReason = 'inherited-dependency';
} else {
dep.skipReason = 'unknown-registry';
}
} else {
// Multiple sources that are not all indexes (e.g. a git source
// per platform) cannot be represented as a single update.
dep.skipReason = 'unsupported';
}
} else {
// Dependency is not pinned to a specific source, so we need to
// determine the source based on the index configuration.
if (hasExplicitDefault) {
// don't fall back to pypi if there is an explicit default index
dep.registryUrls = [];
} else if (defaultIndex) {
// There is a default index configured, so use it.
dep.registryUrls = [defaultIndex.url];
}
if (implicitIndexUrls?.length) {
// If there are implicit indexes, check them first and fall back
// to the default.
dep.registryUrls = implicitIndexUrls.concat(
dep.registryUrls ?? PypiDatasource.defaultURL,
);
}
}
}
}
return deps;
}
async extractLockedVersions(
project: PyProject,
deps: PackageDependency[],
packageFile: string,
): Promise<PackageDependency[]> {
const lockFileName = await findLocalSiblingOrParent(
packageFile,
this.lockfileName,
);
if (lockFileName === null) {
logger.debug({ packageFile }, `No uv lock file found`);
} else {
const lockFileContent = await readLocalFile(lockFileName, 'utf8');
if (lockFileContent) {
const { val: lockFileMapping, err } = Result.parse(
lockFileContent,
UvLockfile,
).unwrap();
if (err) {
logger.debug({ packageFile, err }, `Error parsing uv lock file`);
} else {
for (const dep of deps) {
const packageName = dep.packageName;
if (packageName && packageName in lockFileMapping) {
dep.lockedVersion = lockFileMapping[packageName];
}
}
}
}
}
return Promise.resolve(deps);
}
async updateArtifacts(
updateArtifact: UpdateArtifact,
project: PyProject,
): Promise<UpdateArtifactsResult[] | null> {
const { config, updatedDeps, packageFileName } = updateArtifact;
const { isLockFileMaintenance } = config;
// abort if no lockfile is defined
const lockFileName = await findLocalSiblingOrParent(
packageFileName,
'uv.lock',
);
if (lockFileName === null) {
logger.debug({ packageFileName }, `No uv lock file found`);
return null;
}
try {
const existingLockFileContent = await readLocalFile(lockFileName, 'utf8');
if (!existingLockFileContent) {
logger.debug('No uv.lock found');
return null;
}
const pythonConstraint: ToolConstraint = {
toolName: 'python',
constraint:
config.constraints?.python ?? project.project?.['requires-python'],
};
const uvConstraint: ToolConstraint = {
toolName: 'uv',
constraint:
config.constraints?.uv ?? project.tool?.uv?.['required-version'],
};
const extraEnv = {
...(await getUvExtraIndexUrl(project, updateArtifact.updatedDeps)),
...(await getUvIndexCredentials(project)),
};
const execOptions: ExecOptions = {
cwdFile: packageFileName,
extraEnv,
docker: {},
toolConstraints: [pythonConstraint, uvConstraint],
};
// on lockFileMaintenance do not specify any packages and update the complete lock file
// else only update specific packages
let cmd: string;
if (isLockFileMaintenance) {
cmd = `${uvUpdateCMD} --upgrade`;
} else {
cmd = generateCMD(updatedDeps);
}
await gitExec(cmd, execOptions);
// check for changes
const fileChanges: UpdateArtifactsResult[] = [];
const newLockContent = await readLocalFile(lockFileName, 'utf8');
const isLockFileChanged = existingLockFileContent !== newLockContent;
if (isLockFileChanged) {
fileChanges.push({
file: {
type: 'addition',
path: lockFileName,
contents: newLockContent,
},
});
} else {
logger.debug('uv.lock is unchanged');
}
return fileChanges.length ? fileChanges : null;
} catch (err) {
if (err.message === TEMPORARY_ERROR) {
throw err;
}
logger.debug({ err }, 'Failed to update uv lock file');
return [
{
artifactError: {
fileName: lockFileName,
stderr: err.message,
},
},
];
}
}
}
function generateCMD(updatedDeps: Upgrade[]): string {
const deps: string[] = [];
for (const dep of updatedDeps) {
switch (dep.depType) {
case depTypes.optionalDependencies: {
deps.push(dep.depName!);
break;
}
case depTypes.uvDevDependencies:
case depTypes.uvSources: {
deps.push(dep.depName!);
break;
}
case depTypes.buildSystemRequires:
// build requirements are not locked in the lock files, no need to update.
break;
default: {
deps.push(dep.packageName!);
}
}
}
return `${uvUpdateCMD} ${deps.map((dep) => `--upgrade-package ${quote(dep)}`).join(' ')}`;
}
function getMatchingHostRule(url: string | undefined): HostRule {
return find({ hostType: PypiDatasource.id, url });
}
async function getUsernamePassword(
url: URL,
): Promise<{ username?: string; password?: string }> {
const rule = getMatchingHostRule(url.toString());
if (rule.username || rule.password) {
return rule;
}
if (url.hostname.endsWith('.pkg.dev')) {
const hostRule = await getGoogleAuthHostRule();
if (hostRule) {
return hostRule;
}
logger.once.debug({ url }, 'Could not get Google access token');
}
return {};
}
async function getUvExtraIndexUrl(
project: PyProject,
deps: Upgrade[],
): Promise<NodeJS.ProcessEnv> {
const pyPiRegistryUrls = deps
.filter((dep) => dep.datasource === PypiDatasource.id)
.filter((dep) => {
// Remove dependencies that are pinned to a specific index
const sources = project.tool?.uv?.sources;
const packageName = dep.packageName!;
return !sources || !(packageName in sources);
})
.flatMap((dep) => dep.registryUrls)
.filter(isString)
.filter((registryUrl) => {
// Check if the registry URL is not the default one and not already configured
const configuredIndexUrls = coerceArray(
project.tool?.uv?.index?.map(({ url }) => url),
);
return (
registryUrl !== PypiDatasource.defaultURL &&
!configuredIndexUrls.includes(registryUrl)
);
});
const registryUrls = new Set(pyPiRegistryUrls);
const extraIndexUrls: string[] = [];
for (const registryUrl of registryUrls) {
const parsedUrl = parseUrl(registryUrl);
if (!parsedUrl) {
continue;
}
const { username, password } = await getUsernamePassword(parsedUrl);
if (username || password) {
if (username) {
parsedUrl.username = username;
}
if (password) {
parsedUrl.password = password;
}
}
extraIndexUrls.push(parsedUrl.toString());
}
return {
UV_EXTRA_INDEX_URL: extraIndexUrls.join(' '),
};
}
async function getUvIndexCredentials(
project: PyProject,
): Promise<NodeJS.ProcessEnv> {
const uv_indexes = project.tool?.uv?.index;
if (!uv_indexes) {
return {};
}
const entries = [];
for (const { name, url } of uv_indexes) {
const parsedUrl = parseUrl(url);
/* v8 ignore next -- needs test */
if (!parsedUrl) {
continue;
}
// If no name is provided for the index, authentication information must be passed through alternative methods
if (!name) {
continue;
}
const { username, password } = await getUsernamePassword(parsedUrl);
const NAME = name.toUpperCase().replace(regEx(/[^A-Z0-9]/g), '_');
if (username) {
entries.push([`UV_INDEX_${NAME}_USERNAME`, username]);
}
if (password) {
entries.push([`UV_INDEX_${NAME}_PASSWORD`, password]);
}
}
return Object.fromEntries(entries);
}