Mamba API example

mamba_api_usage.py

import os

from mamba import mamba_api as api
from conda.models.channel import Channel, prioritize_channels
from conda.core.index import calculate_channel_urls, check_whitelist
from conda.core.subdir_data import cache_fn_url, create_cache_dir


def get_index(channel_urls=(), prepend=True, platform=None,
              use_local=False, use_cache=False, unknown=None, prefix=None,
              repodata_fn="repodata.json"):
    real_urls = calculate_channel_urls(channel_urls, prepend, platform, use_local)
    check_whitelist(real_urls)

    dlist = api.DownloadTargetList()

    sddata = []
    index = []
    for idx, url in enumerate(real_urls):
        channel = Channel(url)

        full_url = channel.url(with_credentials=True) + '/' + repodata_fn
        full_path_cache = os.path.join(
            create_cache_dir(),
            cache_fn_url(full_url, repodata_fn))

        sd = api.SubdirData(channel.name + '/' + channel.subdir,
                            full_url,
                            full_path_cache)

        sd.load()
        index.append((sd, channel))
        dlist.add(sd)

    is_downloaded = dlist.download(True)

    if not is_downloaded:
        raise RuntimeError("Error downloading repodata.")

    return index


class MambaSolver:

    def __init__(self, channels):
        index = get_index(channels)

        self.pool = api.Pool()
        self.repos = []

        priority = 0
        subpriority = 0  # wrong! :)
        for subdir, channel in index:
            repo = api.Repo(self.pool, str(channel), subdir.cache_path(), channel.url(with_credentials=True))
            repo.set_priority(priority, subpriority)
            self.repos.append(repo)


    def solve(self, specs):
        solver_options = [(api.SOLVER_FLAG_ALLOW_DOWNGRADE, 1)]
        solver = api.Solver(self.pool, solver_options)

        # normalization can be performed using Conda's MatchSpec
        solver.add_jobs(specs, api.SOLVER_INSTALL)
        success = solver.solve()
        
        if not success:
            # failed!
            print(solver.problems_to_str())

        transaction = api.Transaction(solver)
        transaction.print()

        to_install, to_remove = transaction.to_conda()

        # to_install is a list of tuples with 3 elements:
        # element 0 -> repository URL
        # element 1 -> filename of the package to install
        # element 2 -> JSON string of repodata information

        # to_remove is list of tuple with 2 elements
        # element 0 -> repo name (should always be "installed")
        # element 1 -> filename

        print("To be installed\n================")
        print('\n'.join([x[1] for x in to_install]))
        print("To be removed\n================")
        print('\n'.join([x[1] for x in to_remove]))

if __name__ == "__main__":
    solver = MambaSolver(['conda-forge/linux-64', 'conda-forge/noarch'])
    solver.solve(["xtensor 0.18"])