True used as superuser name, otherwise user to connect as to the database denotes by db_* parameter of this function
@plac.annotations(
osm_files=("a comma (`,`) separated list of OSM files to be passed to osm2pgsql (gunzipped files are accepted if osm2pgsql accepts them (the version installed by osm_postgis_transform_prequisites does))", "positional", None, a_list),
skip_start_db=("Specify this flag in order to feed the data to an already running postgres process which the script will attempt to connect to with the parameters specified by `db-host`, `db-port`, `db-user`, `db-password` and `db-name` arguments.", "flag"),
data_dir=("The directory which contains or will contain the data of the PostGIS database (see documentation of `-D` option in `man initdb` for further details). The directory will be created if it doesn't exist. If a file is passed as argument, the script will fail argument validation. The script will fail if the directory is an invalid PostGIS data directory (e.g. one which allows partial start of a `postgres` process but contains invalid permissions or misses files). As soon as a non-empty directory is passed as argument, it is expected to be a valid PostGIS data directory! If the script fails due to an unexpected error, YOU have to take care of cleaning that directory from anything besides the stuff inside before the script has been invoked!", "option"),
db_host=("The host where the nested database process should run (has to be specified if default value isn't reachable) or the host where to reach the already running postgres process (see --start-db for details)", "option"),
db_port=("The port where the nested database process will be listening (has to be specified if the port denoted by the default value is occupied) or the port where to reach the already running postgres process (see --start-db for details)", "option"),
db_user=("name of user to use for authentication at the database (will be created if database doesn't exist) (see --start-db for details)", "option"),
db_password=("password for the user specified with `--db-user` argument to use for authentication at the database (will be set up if database doesn't exist) (see --start-db for details)", "option"),
db_name=("name of the database to connect to or to be created (see --start-db for details)", "option"),
cache_size=("size of osm2pgsql cache (see `--cache` option of `man osm2pgsql`)", "option"),
osm2pgsql=("optional path to a osm2pgsql binary", "option"),
)
def osm_postgis_transform(osm_files, skip_start_db, data_dir=data_dir_default, db_host=db_host_default, db_port=db_port_default, db_user=db_user_default, db_password=db_password_default, db_name=db_name_default, cache_size=cache_size_default, osm2pgsql=osm2pgsql_default):
# the text for the help transformed by plac:
"""
This script sets up PostGIS database with data from an OSM (.osm) file. It
is essentially a wrapper around `osm2pgsql`. By default it will either spawn a database process based on the data directory speified with the `--data-dir` argument (if the data directory is non-empty) or create a database data directory and spawn a database process based on that newly created data directory and feed data to it. If the nested database process can't be connected to with the default value for database connection parameters, they have to be overwritten, otherwise the script will fail with the error message of the `postgres` process.
The start of a nested database process can be skipped if `--skip-start-db` command line flag is set. In this case the database connection parameters will be used to connect to an external already running `postgres` process where data will be fed to.
WARNING: The script has not yet been tested completely to hide database credentials (including the password) from output and/or other logging backends (files, syslog, etc.). It is currently recommended to specify a separate database and local host for the script only and to not care about it at all (as OSM data is as far from a secret as it could be).
"""
if osm_files is None:
raise ValueError("osm_files mustn't be None")
if str(type(osm_files)) != "file_ if it doesn't exist
def retrieve_column_values(output, column_count, comment_symbol="#"):
output_lines0 = filter_output_lines(output, comment_symbol=comment_symbol)
ret_value = []
for output_line in output_lines0:
column_value = retrieve_column_from_line(output_line, column_count)
ret_value.append(column_value)
return ret_value
# reads file_ and returns the return value of output_lines invoked with the read file content and the specified comment_symbol
# @args comment_symbol can be None in order to include all lines, must not be the empty string '' (ValueError will be raised)
def file_lines(file_, comment_symbol="#"):
if file_ == None:
raise ValueError("file_ mustn't be None")
if comment_symbol == "":
raise ValueError("comment_symbol mustn't be the empty string ''")
file_obj = open(file_,"r")
file_content = file_obj.read()
file_obj.close()
file_lines = file_content.split("\n")
ret_value = filter_output_lines(file_lines, comment_symbol)
return ret_value
# removes all empty lines and lines which start with comment_symbol (after eventual whitespace) from lines which is supposed to be splitted content of a file or splitted output of a command
# @args comment_symbol can be None in order to include all lines, must not be the empty string '' (ValueError will be raised)
def filter_output_lines(lines, comment_symbol="#"):
if comment_symbol == "":
raise ValueError("comment_symbol mustn't be the empty string ''")
if str(type(lines)) != "None in order to include all lines, must not be the empty string '' (ValueError will be raised)
# @return a list with all line which match pattern or an empty list if none matches pattern, never None
# @see file_lines_match for a fail-fast behavior which just indicates whether a matching line has been found using a boolean
def file_lines_matches(file_,pattern,comment_symbol="#"):
if comment_symbol == "":
raise ValueError("comment_symbol mustn't be the empty string ''")
retvalue = []
filelines = file_lines(file_,comment_symbol) # comment lines are already skipped here
return output_lines_matches(filelines,pattern,comment_symbol=comment_symbol)
def output_lines_matches(lines, pattern, comment_symbol="#"):
retvalue = []
lines = filter_output_lines(lines, comment_symbol=comment_symbol) # remove comment lines and
for line in lines:
if re.match(pattern,line) != None:
retvalue.append(line)
return retvalue
# checks whether at least one (non-commented) line matches pattern and returns True if this is the case. The method does the same like file_lines_matches with fail-fast behavior.
# @args comment_symbol can be None in order to include all lines, must not be the empty string '' (ValueError will be raised)
# @return True if at least one line matches pattern, False otherwise
def file_lines_match(file_, pattern, comment_symbol="#"):
if comment_symbol == "":
raise ValueError("comment_symbol mustn't be the empty string ''")
file_lines0 = file_lines(file_, comment_symbol=comment_symbol) # comment lines are already skipped here
return output_lines_match(file_lines0, pattern, comment_symbol=comment_symbol)
# @args comment_symbol can be None in order to include all lines, must not be the empty string '' (ValueError will be raised)
def output_lines_match(lines, pattern, comment_symbol="#"):
if comment_symbol == "":
raise ValueError("comment_symbol mustn't be the empty string ''")
lines = filter_output_lines(lines,comment_symbol=comment_symbol)
for line in lines:
if re.match(pattern, line):
return True
return False
# @args line the line to be commented out (can be a regular expression or a literal) (leading and trailing whitespace in lines in the file will be ignored)
# @args comment_symbol can be None in order to include all lines, must not be the empty string '' (ValueError will be raised)
def comment_out(file_path, line, comment_symbol):
if comment_symbol == "":
raise ValueError("comment_symbol mustn't be the empty string ''")
new_lines = []
file_lines0 = file_lines(file_path, comment_symbol=None)
for file_line in file_lines0:
if not re.match("[\\s]*%s[\\s]*" % line):
new_lines.append(file_line)
else:
new_lines.append("%s %s" % (comment_symbol, line))
file_obj = open(file_path, "rw+")
for new_line in new_lines:
file_obj.write("%s\n" % new_line)
file_obj.flush()
file_obj.close()
# @args line the line to be commented in (can be a regular expression or a literal)
def comment_in(file_path, line, comment_symbol):
new_lines = []
file_lines0 = file_lines(file_path, comment_symbol=None)
for file_line in file_lines0:
if not re.match("[\\s]*%s[\\s]*%s" % (comment_symbol, line), line):
new_lines.append(file_line)
else:
new_lines.append(re.search(line, file_line).group(0))
file_obj = open(file_path, "w")
for new_line in new_lines:
file_obj.write("%s\n" % new_line)
file_obj.flush()
file_obj.close()
def create_file_wrapper(path):
if not check_os.check_python3():
file_obj = open(path, "w") # opening for writing or appending creates a non-existant file
file_obj.close()
else:
file_obj = open(path, "x")
file_obj.close()
PK H"EܽB ( ( # python_essentials/lib/file_utils.py#!/usr/bin/python
# -*- coding: utf-8 -*-
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see True.
# @args backup creates a backup in the form filename.bk where N is the minimal number denoting a filesname which doesn't exist. Backup is not created if the content of point_to and link_name are identical. If link_name is a link, the link target is backed up (because backing up the link doesn't make sense).
# @return False if the link could not be created due to existance of the link_name (and force being False).
def checked_link(point_to, link_name, force=False, backup_type=backup_type_default):
if not backup_type in backup_types:
raise ValueError("backup_type '%s' isn't one of %s" % (backup_type,backup_types))
target_parent = os.path.abspath(os.path.join(link_name, '..'))
if not os.path.lexists(target_parent):
os.makedirs(target_parent)
if not os.path.lexists(link_name):
os.symlink(point_to,link_name)
return True
else:
if not force:
return False
if os.path.islink(link_name) and os.path.realpath(link_name) == point_to:
return True
if os.path.exists(link_name):
if backup_type != BACKUP_TYPE_NONE:
backup_file(link_name, backup_type=backup_type)
else:
if not os.path.isdir(link_name) or os.path.islink(link_name):
os.remove(link_name)
else:
shutil.rmtree(link_name)
os.symlink(point_to, link_name)
return True
# implementation notes:
# - skipping backup if file content are identical is not useful because if was initially intended to avoid endless identical backups of links which now handled smarter by skipping link creation when link exists and points to the source already
################################################################################
# Write tools #
################################################################################
def write_file(file0, what):
if os.path.isdir(file0):
raise ValueError("file %s is a directory" % file0)
if lazy_newline and not what.endswith("\n"):
what = "%s\n" % what
file_obj = open(file0, 'w')
file_obj.write(what)
file_obj.flush()
file_obj.close()
# appends what to file with path file0. Creates file if it doesn't exist. Newline character is appended if lazy_newline is True and what doesn't end with a newline yet.
# raises ValueError if file0 is a directory
def append_file(file0, what, lazy_newline=True):
if os.path.isdir(file0):
raise ValueError("file %s is a directory" % file0)
if lazy_newline and not what.endswith("\n"):
what = "%s\n" % what
file_obj = open(file0, 'a')
file_obj.write(what)
file_obj.flush()
file_obj.close()
def check_dir(dir_path):
if not os.path.exists(dir_path):
return False
if not os.path.isdir(dir_path):
raise ValueError("dir_path has to point to a directory")
if len(os.listdir(dir_path)) == 0:
logger.debug("%s is empty" % dir_path)
return False
return True
def create_dir(dir_path, allow_content):
if os.path.exists(dir_path):
if os.path.isdir(dir_path):
if len(os.listdir(dir_path)) > 0 and not allow_content:
raise RuntimeError("%s exists and has content, a directory is supposed to be created. Remove or move the content of the directory" % dir_path)
else:
# file
raise RuntimeError("%s exists, a directory is supposed to be created. Remove or move the file" % dir_path)
else:
os.makedirs(dir_path)
def check_file(file_path, md5sum):
if not os.path.exists(file_path):
return False
if not os.path.isfile(file_path):
raise ValueError("file_path has to point to a file")
retrieved_md5sum = retrieve_hash(file_path)
if not retrieved_md5sum == md5sum:
logger.debug("%s has md5 sum %s instead of %s" % (file_path, retrieved_md5sum, md5sum))
return False
return True
def retrieve_hash(file_path, hasher="md5", blocksize=65536):
afile = open(file_path, "r")
hasher_instance = hashlib.new(hasher)
buf = afile.read(blocksize)
while len(buf) > 0:
hasher_instance.update(buf)
buf = afile.read(blocksize)
afile.close()
ret_value = hasher_instance.hexdigest()
return ret_value
# @return the XOR of all file content, name and directory name hashes in order
# to track any changes (even one simple rename of a file or directory)
def retrieve_hash_recursive(dir_path, hasher="md5", blocksize=65536):
logger.debug("generating recursive hash for '%s'" % (dir_path,))
if not os.path.isdir(dir_path):
raise ValueError("dir_path '%s' isn't a directory" % (dir_path,))
ret_value = 0
for dirpath, dirnames, filenames in os.walk(dir_path):
for filename in filenames:
current_file_path = os.path.join(dirpath, filename)
hasher_instance = hashlib.new(hasher)
hasher_instance.update(current_file_path)
ret_value = ret_value ^ int(hasher_instance.hexdigest(), base=16)
ret_value = ret_value ^ int(retrieve_hash(current_file_path, hasher=hasher, blocksize=65536), base=16)
for dirname in dirnames:
current_dir_path = os.path.join(dirpath, dirname)
hasher_instance = hashlib.new(hasher)
hasher_instance.update(current_dir_path)
ret_value = ret_value ^ int(hasher_instance.hexdigest(), base=16)
return str(ret_value)
# moves the file to a backup location (i.e. file_path no longer exists after
# the function run)
# @args file_path a path of an existing file or directory
def backup_file(file_path, backup_type=backup_type_default, follow_links=False):
if not os.path.exists(file_path):
raise ValueError("file_path '%s' doesn't exist, backup_file only accepts an existing file or directory as argument" % (file_path,))
if backup_type == BACKUP_TYPE_SHA1:
if os.path.isdir(file_path):
file_sha1 = retrieve_hash_recursive(file_path, hasher="sha1")
else:
file_sha1 = retrieve_hash(file_path, hasher="sha1")
file_backup_path = "%s.bk-%s" % (file_path, file_sha1)
if not os.path.exists(file_backup_path):
os.rename(os.path.realpath(file_path), file_backup_path)
elif backup_type == BACKUP_TYPE_SEQUENCE:
renamepostfix = ".bk"
renamepostfixcount = 0
backup_file = file_path
if follow_links:
backup_file = os.path.realpath(file_path)
if os.path.lexists(backup_file):
renamepostfix = ".bk%d" % renamepostfixcount
renamepostfixcount += 1
while os.path.lexists(backup_file+renamepostfix):
renamepostfix = ".bk%d" % renamepostfixcount
renamepostfixcount += 1
target_backup_path = backup_file+renamepostfix
logger.debug("backing up %s to %s" % (backup_file, target_backup_path))
os.rename(backup_file, target_backup_path)
else:
raise ValueError("backup_type '%s' isn't supported" % (backup_type,))
PK S
HFϒXtv! v! $ python_essentials/lib/mount_utils.py#!/usr/bin/python
# coding: utf-8
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see id -u -n , but this is buggy in OpenSUSE due to missing services (as so many other commands)
def username_by_id(uid):
if str(type(uid)) != "":
raise ValueError("uid has to be an int")
passwd_lines = file_line_utils.file_lines("/etc/passwd", "#")
for passwd_line in passwd_lines:
passwd_line_content = passwd_line.split(":")
if int(passwd_line_content[2]) == uid:
return passwd_line_content[0]
return None
def groupname_by_id(gid):
if str(type(uid)) != "":
raise ValueError("uid has to be an int")
passwd_lines = file_line_utils.file_files("/etc/passwd", "#")
for passwd_line in passwd_lines:
passwd_line_content = passwd_line.split(":")
if (passwd_line_content[3]) == gid:
return passwd_line_content[0]
return None
# @return the effective gid or -1 if group groupname doesn't exist
def id_by_username(username):
if not check_user_exists(username):
return -1
ret_value = subprocess.check_output(["id", "-u", username])
return int(ret_value)
# @return the effective gid or -1 if group groupname doesn't exist
def id_by_groupname(groupname):
if not check_group_exists(groupname):
return -1
group_lines = file_line_utils.file_lines("/etc/group", comment_symbol="#")
for group_line in group_lines:
group_line_content = group_line.split(":")
if (group_line_content[0]) == groupname:
return int(group_line_content[2])
return None
# implementation notes:
# - id -g username is simply wrong
# doesn't handle lines which start with whitespace in /etc/passwd correctly
# @return True if username exists (in /etc/passwd)
def check_user_exists(username):
passwd_lines = file_line_utils.file_lines("/etc/passwd", comment_symbol="#")
for passwd_line in passwd_lines:
if passwd_line.startswith(username):
return True
return False
# doesn't handle lines which start with whitespace in /etc/group correctly
# @return True if groupname exists (in /etc/group)
def check_group_exists(groupname):
group_lines = file_line_utils.file_lines("/etc/group", comment_symbol="#")
for group_line in group_lines:
if group_line.startswith(groupname):
return True
return False
# to be passed to preexec_fn argument of relevant subprocess.* functions, e.g.
# return_code = sp.check_call(["git", "rev-parse"], stderr=sp.PIPE, cwd=base_dir, preexec_fn=user_group_utils.demote_uid(user_group_utils.id_by_username(build_user)))
#
def demote_uid(uid):
return demote_uid_gid(uid,uid)
def demote_uid_gid(uid, gid):
def ret_value():
os.setgid(gid)
os.setuid(uid)
return ret_value
PK G bS S ! python_essentials/lib/check_os.py#!/usr/bin/python
# -*- coding: utf-8 -*-
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see True if the python version is >= 3.0, otherwise False
def check_python3():
ret_value = sys.version_info >= (3,0)
return ret_value
def findout_architecture():
architecture = sp.check_output(["uname","-m"]).strip()
return architecture
def findout_instruction_set():
architecture = findout_architecture()
instruction_set = ARCHITECTURE_INSTRUCTION_SET_DICT[architecture]
return instruction_set
#lsb_release only works with python2.x
def findout_release_ubuntu():
while not os.path.isfile("/usr/bin/python2.6") and not os.path.isfile("/usr/bin/python2.7"):
print("Neither python2.6 nor python 2.7 could be found in /usr/bin/. It's necessary to determine your distribution release")
confirm("proceed","Install python 2.6 or 2.7 and make it available at /usr/bin/python2.6 or /usr/lib/python2.7")
release= sp.check_output([lsb_release,"-cs"]).strip().decode("utf-8")
return release
def findout_release_debian():
return findout_release_ubuntu()
# useful is a feature is available for any version up from a certain (the tuple contains ints because strings are less comparable)
def findout_release_ubuntu_tuple():
while not os.path.isfile("/usr/bin/python2.6") and not os.path.isfile("/usr/bin/python2.7"):
print("Neither python2.6 nor python 2.7 could be found in /usr/bin/. It's necessary to determine your distribution release")
confirm("proceed","Install python 2.6 or 2.7 and make it available at /usr/bin/python2.6 or /usr/lib/python2.7")
release_number = sp.check_output([lsb_release, "-rs"]).strip().decode("utf-8")
release_tuple = tuple([int(x) for x in release_number.split(".")])
return release_tuple
# linuxmint release identifiers are natural numbers
def findout_release_linuxmint():
ret_value = sp.check_output([lsb_release, "-r", "-s"]).decode("utf-8").strip()
return int(ret_value)
PK VG<\C C ! python_essentials/lib/pm_utils.py#!/usr/bin/python
# -*- coding: utf-8 -*-
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see True if package_name is installed, False otherwise
def dpkg_check_package_installed(package_name):
# old implementation relying on dpkg return codes (reported https://bugs.launchpad.net/ubuntu/+source/dpkg/+bug/1380326 for clearification of them) (more elegant, but in Ubuntu 14.10-beta1 return code is 0 for both installed and uninstalled packages; working dpkg of version 1.17.13ubuntu1 in Ubuntu 14.10 is the same like in Ubuntu 14.10-beta1 -> assume unexplainable anomaly, abstract implementation and make it possible to choose between returncode and output based implementation (not too elegant, but intuitive))
def __dpkg_check_package_installed_returncode__():
return_code = sp.call(["dpkg", "-s", package_name], stdout=sp.PIPE, stderr=sp.PIPE)
return return_code == 0
def __dpkg_check_package_installed_output__():
dpkg_output = sp.check_output(["dpkg", "-s", package_name])
ret_value = "Package: %s\nStatus: install ok installed" % (package_name,) in dpkg_output
return ret_value
ret_value = __dpkg_check_package_installed_returncode__()
return ret_value
##############################
# apt-get tools
##############################
# a wrapper around apt-get dist-upgrade to use the internal aptuptodate flag
def upgrade(package_manager=PACKAGE_MANAGER_APT_GET , assume_yes=assume_yes_default, skip_apt_update=skip_apt_update_default, install_recommends=install_recommends_default, install_suggests=install_suggests_default):
if package_manager == PACKAGE_MANAGER_APT_GET:
aptupdate(skip_apt_update)
command_list = [apt_get, "dist-upgrade"]
options_command_list = __generate_apt_options_command_list__(assume_yes=assume_yes, install_recommends=install_recommends, install_suggests=install_suggests)
sp.check_call(command_list+options_command_list)
else:
raise RuntimeError("package_manager %s not yet supported" % (package_manager,))
def install_apt_get_build_dep(packages, package_manager="apt-get", assume_yes=assume_yes_default, skip_apt_update=skip_apt_update_default):
if packages == None or not type(packages) == type([]):
raise Exception("packages has to be not None and a list")
if len(packages) == 0:
return 0
aptupdate(skip_apt_update)
for package in packages:
apt_get_output = sp.check_output([apt_get, "--dry-run", "build-dep", package]).strip()
apt_get_output_lines = apt_get_output.split("\n")
build_dep_packages = []
for apt_get_output_line in apt_get_output_lines:
if apt_get_output_line.startswith(" "):
build_dep_packages += re.split("[\\s]+", apt_get_output_line)
build_dep_packages = [x for x in build_dep_packages if x != ""]
install_packages(build_dep_packages, package_manager, assume_yes, skip_apt_update=skip_apt_update)
# only checks for the specified packages, no for their recommends or suggests
# @return True if all packages in packages are installed via package_manager, False otherwise
def check_packages_installed(packages, package_manager="apt-get", skip_apt_update=skip_apt_update_default):
package_managers = ["apt-get"]
if package_manager == "apt-get":
for package in packages:
package_installed = dpkg_check_package_installed(package)
if not package_installed:
return False
return True
else:
raise Exception("package_manager has to be one of "+str(package_managers))
# internal implementation notes:
# - python-apt bindings have been dropped due to slow speed (dpkg queries are much faster)
# quiet flag doesn't make sense because update can't be performed quietly obviously (maybe consider to switch to apt-api)
# return apt-get return code or 0 if packages are already installed or packages is empty
def install_packages(packages, package_manager="apt-get", assume_yes=assume_yes_default, skip_apt_update=skip_apt_update_default, install_recommends=install_recommends_default, install_suggests=install_suggests_default, preexec_fn=None):
if check_packages_installed(packages, package_manager, skip_apt_update=skip_apt_update):
return 0
return __package_manager_action__(packages, package_manager, ["install"], assume_yes, skip_apt_update=skip_apt_update, install_recommends=install_recommends, install_suggests=install_suggests, preexec_fn=preexec_fn)
# doesn't check whether packages are installed
# @return apt-get return code or 0 if packages is empty
def reinstall_packages(packages, package_manager="apt-get", assume_yes=assume_yes_default, skip_apt_update=skip_apt_update_default, stdout=None, preexec_fn=None):
return __package_manager_action__(packages, package_manager, ["--reinstall", "install"], assume_yes, skip_apt_update=skip_apt_update,stdout=None, preexec_fn=preexec_fn)
# @return apt-get return code oder 0 if none in packages is installed or packages is empty
def remove_packages(packages, package_manager="apt-get", assume_yes=assume_yes_default, skip_apt_update=skip_apt_update_default, preexec_fn=None):
return __package_manager_action__(packages, package_manager, ["remove"], assume_yes, skip_apt_update=skip_apt_update, preexec_fn=preexec_fn)
def __generate_apt_options_command_list__(assume_yes=assume_yes_default, install_recommends=install_recommends_default, install_suggests=install_suggests_default):
# apt-get installs recommended packages by default, therefore the
# option to deactivate it is negative (--no-install-recommends), while
# the option to install suggests is option, there the option is positive
command_list = []
if not install_recommends:
command_list.append("--no-install-recommends")
if install_suggests:
command_list.append("--install-suggests")
if assume_yes:
command_list.append("--assume-yes")
return command_list
def __package_manager_action__(packages, package_manager, package_manager_action, assume_yes, skip_apt_update=skip_apt_update_default, stdout=None, install_recommends=install_recommends_default, install_suggests=install_suggests_default, preexec_fn=None):
"""quiet flag doesn't make sense because update can't be performed quietly obviously (maybe consider to switch to apt-api)
@args packages a list of command to be inserted after the package manager command and default options and before the package list
@args preexec_fn a function to be passed to the `preexec_fn` argument of `subprocess.Popen`
"""
if not "" == str(type(packages)) and str(type(packages)) != "":
raise ValueError("packages isn't a list")
if len(packages) == 0:
return 0
if package_manager == "apt-get":
aptupdate(skip_apt_update)
command_list = [apt_get]
options_command_list = __generate_apt_options_command_list__(assume_yes=assume_yes, install_recommends=install_recommends, install_suggests=install_suggests)
sp.check_call(command_list+options_command_list+package_manager_action+packages, preexec_fn=preexec_fn)
elif package_manager == "yast2":
sp.check_call(["/sbin/yast2", "--"+package_manager_action]+packages, preexec_fn=preexec_fn) # yast2 doesn't accept string concatenation of packages with blank, but the passed list (it's acutually better style...)
elif package_manager == "zypper":
sp.check_call(["zypper", package_manager_action]+packages, preexec_fn=preexec_fn)
elif package_manager == "equo":
sp.check_call(["equo", package_manager_action]+packages, preexec_fn=preexec_fn)
else:
raise ValueError(str(package_manager)+" is not a supported package manager")
# implementation notes:
# - don't return catched exceptions, but just throw them (that's what exceptions are for)
# - not a good idea to redirect output of subcommand other than update to file because interaction (e.g. choosing default display manager, comparing config file versions) is useful and necessary and suppressed when redirecting to file
# - checking availability of apt lock programmatically causes incompatibility with invokation without root privileges -> removed
# updates apt using apt-get update command. Throws exceptions as specified by sp.check_call if the command fails
def invalidate_apt():
logger.debug("invalidating apt status (update forced at next package manager action)")
global apt_invalid
apt_invalid = True
global aptuptodate
aptuptodate = False
# updates apt using sp.check_call, i.e. caller has to take care to handle exceptions which happen during execution
def aptupdate(skip=skip_apt_update_default, force=False):
global aptuptodate
if (not aptuptodate and not skip) or force or apt_invalid:
print("updating apt sources")
apt_stdout = None
if APT_OUTPUT == APT_OUTPUT_TMP_FILE:
apt_get_update_log_file_tuple = tempfile.mkstemp("libinstall_apt_get_update.log")
logger.info ("logging output of apt-get update to %s" % apt_get_update_log_file_tuple[1])
apt_stdout = apt_get_update_log_file_tuple[0]
sp.check_call([apt_get, "--quiet", "update"], stdout=apt_stdout)
aptuptodate = True
################################################################################
# apt source entries #
################################################################################
valid_source_line_types = ["deb", "deb-src"]
def __validate_apt_source_function_params_augeas_root__(augeas_root, sources_dir_path, sources_file_path):
if augeas_root is None:
raise ValueError("augeas_root mustn't be None")
if not os.path.exists(augeas_root):
raise ValueError("augeas_root '%s' doesn't exist" % (augeas_root,))
if not os.path.isdir(augeas_root):
raise ValueError("augeas_root '%s' isn't a directory, but has to be" %
(augeas_root))
if not os.path.exists(sources_dir_path):
raise ValueError("sources_dir_path '%s' doesn't exist" %
(sources_dir_path,))
if not os.path.exists(sources_file_path):
raise ValueError("sources_file_path '%s' doesn't exist" %
(sources_file_path,))
def __validate_apt_source_function_params_type__(the_type):
if not the_type in valid_source_line_types:
raise ValueError("the_type '%s' isn't a valid source line type (has to be one of %s)" % (the_type, valid_source_line_types))
# Avoids the weakness of add-apt-repository command to add commented duplicates of lines which are already present by not adding those at all.
# @args uri the URI of the apt line
# @args component the component to be served (e.g. main)
# @args distribution the distribution of the entry (e.g. trusty for an Ubuntu 14.04 system)
# @args the_type the type of the entry (usually deb or deb-src)
def check_apt_source_line_added(uri, component, distribution, the_type, augeas_root="/",):
sources_dir_path= os.path.join(augeas_root, "etc/apt/sources.list.d")
sources_file_path= os.path.join(augeas_root, "etc/apt/sources.list")
__validate_apt_source_function_params_augeas_root__(augeas_root, sources_dir_path, sources_file_path)
__validate_apt_source_function_params_type__(the_type)
a = augeas.Augeas(root=augeas_root)
# workaround missing loop label feature with inner function
def __search__():
for sources_dir_file in [os.path.join(sources_dir_path, x) for x in os.listdir(sources_dir_path)]:
commented_in_lines = a.match("/files/%s/*" % (os.path.relpath(sources_dir_file,augeas_root),)) # match doesn't return lines with comments; it doesn't matter whether ppa_sources_d_file starts with / for the match statement
for commented_in_line in commented_in_lines:
if a.get("%s/uri" % (commented_in_line,)) == uri and a.get("%s/component" % (commented_in_line,)) == component and a.get("%s/distribution" % (commented_in_line,)) == distribution and a.get("%s/type" % (commented_in_line,)) == the_type:
return True
return False
match_found = __search__()
if match_found:
return True
if not match_found:
# only search in source.list if not found because there's no need to do
# any validation in this this function
commented_in_lines = a.match("/files/%s/*" % (os.path.relpath(sources_file_path,augeas_root),)) # match doesn't return lines with comments; it doesn't matter whether ppa_sources_d_file starts with / for the match statement
match_found = False
for commented_in_line in commented_in_lines:
if a.get("%s/uri" % (commented_in_line,)) == uri and a.get("%s/component" % (commented_in_line,)) == component and a.get("%s/distribution" % (commented_in_line,)) == distribution and a.get("%s/type" % (commented_in_line,)) == the_type:
a.close()
return True
a.close()
return False
# adds an entry based on uri, component,
# distribution and the_type to etc/apt/sources.list
# relatively to augeas_root.
#
# Consider using add-apt-repository or manually adding an entry in a separate
# file into /etc/apt/sources.d/.
#
# There's no validation of parameters except augeas_root, i.e. this
# might mess up your sources.list file quite easily.
def add_apt_source_line(uri, component, distribution, the_type, augeas_root="/",):
sources_dir_path= os.path.join(augeas_root, "etc/apt/sources.list.d")
sources_file_path= os.path.join(augeas_root, "etc/apt/sources.list")
__validate_apt_source_function_params_augeas_root__(augeas_root, sources_dir_path, sources_file_path)
__validate_apt_source_function_params_type__(the_type)
a = augeas.Augeas(root=augeas_root)
a.set("/files/etc/apt/sources.list/01/distribution", distribution) # checkout http://augeas.net/tour.html if you find the 01 label for adding entries confusing (it simply is...)
a.set("/files/etc/apt/sources.list/01/type", the_type)
a.set("/files/etc/apt/sources.list/01/uri", uri)
a.set("/files/etc/apt/sources.list/01/component", component)
a.save()
a.close()
PK VG+ 2 python_essentials/lib/python_essentials_globals.py#!/usr/bin/python
# -*- coding: utf-8 -*-
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see