source: ogLive-Builder-Git/boottools/apt.py @ 2e092b0

#!/usr/bin/python3

import apt
import apt.progress.base
import apt_pkg
import shutil
import os

def clean():
    apt_cache = apt.Cache()

    apt_pkg.init()
    cache_dir = apt_pkg.config.find_dir ('Dir::Cache::archives')

    for fn in os.listdir (cache_dir):
        file_path = os.path.join (cache_dir, fn)
        try:
            if os.path.isfile (file_path):
                os.unlink (file_path)
            elif os.path.isdir (file_path):
                shutil.rmtree (file_path)
        except Exception as e:
            print (f'Failed to delete {file_path}. Reason: {e}')

def autoremove():
    apt_cache = apt.Cache()
    for pkg in apt_cache:
       if pkg.is_installed and pkg.is_auto_removable:
           pkg.mark_delete()
    try:
        apt_cache.commit()
    except Exception as e:
        print ('installation failed: {}'.format (str (e)))

def remove (pkgs):
    apt_cache = apt.Cache()
    for p in pkgs:
        if p in apt_cache:
            apt_cache[p].mark_delete()
    try:
        apt_cache.commit()
    except Exception as e:
        print ('removal failed: {}'.format (str (e)))

def update():
    apt_cache = apt.Cache()
    apt_cache.update()

def upgrade():
    apt_cache = apt.Cache()
    apt_cache.upgrade()

def cache_search (pkgs):
    apt_cache = apt.Cache()
    res = {}
    for p in pkgs:
        res[p] = True if p in apt_cache else False
    return res

def install (pkgs, opts={}):
    apt_pkg.init()

    if opts:
        apt_pkg.init_config()
        for k in opts:
            apt_pkg.config.set (k, opts[k])

    cache = apt_pkg.Cache()
    sl = apt_pkg.SourceList()
    sl.read_main_list()
    cache.update (apt.progress.base.AcquireProgress(), sl)

    _to_install = []
    dep_cache = apt_pkg.DepCache(cache)
    for p in pkgs:
        package = cache[p]
        if not package:
            print (f'package "{p}" not found')
            continue
        _to_install.append (p)
        dep_cache.mark_install(package)

    if _to_install:
        print ('about to install these packages: "{}"'.format (' '.join (_to_install)))
        fetcher = apt_pkg.Acquire()
        install_progress = apt.progress.base.InstallProgress()
        dep_cache.commit(fetcher, install_progress)