Added util.py
Provide a bunch of utilities for soypak. Specify compiled regexes for bottle file parsing. func@ensure_dir: - Will ensure that a directory that exists, and if it doesn't, will create it and any parents. Mode is 0755. - From DNF util module. func@join_path: - Will join any number of string paths together and return a new string path. - From eopkg util module. func@rack_bottle: - Function will aim to write a new bottle file out to the database (nickname: rack). - Either a string representation of a bottle file can be given, or a file descriptor of type TextIOWrapper, can be given as the file pointer: - if a TextIOWrapper, ensure that the pointer is at the start of the file to read the contents again. - if a string, the contents can be set at this. - Detect if there are any comments (useful from translating user defined bottles to be stored - Packages file won't have this). If so, remove them. - Ensure that the database rack exists. - Create the new filename to be stored. - Attempt to open the file in write mode and write the contents to file. Report on the amount of bytes written for clarity. - Function returns true if the file was successfully racked. False upon any failures.
This commit is contained in:
parent
4315780777
commit
7407ba1084
85
soypak/util.py
Normal file
85
soypak/util.py
Normal file
|
|
@ -0,0 +1,85 @@
|
||||||
|
import os
|
||||||
|
import re
|
||||||
|
import clog
|
||||||
|
import errno
|
||||||
|
import soypak.db
|
||||||
|
from io import TextIOWrapper
|
||||||
|
|
||||||
|
|
||||||
|
_log = clog.Logger.get("runtime_logger")
|
||||||
|
|
||||||
|
|
||||||
|
re_comments = re.compile(r"(--.*$|\*--(\n.*?)+--\*)", re.MULTILINE)
|
||||||
|
re_desc_tag = re.compile(r"^Description:(?: .*\n)+", re.MULTILINE)
|
||||||
|
|
||||||
|
|
||||||
|
# from DNF util module (https://github.com/rpm-software-management/dnf/blob/master/dnf/util.py#L142)
|
||||||
|
def ensure_dir(dname) -> tuple[int, str]:
|
||||||
|
try:
|
||||||
|
os.makedirs(dname, mode=0o755)
|
||||||
|
except OSError as err:
|
||||||
|
if err.errno != errno.EEXIST or not os.path.isdir(dname):
|
||||||
|
return err.errno, str(err)
|
||||||
|
|
||||||
|
return 0, ""
|
||||||
|
|
||||||
|
|
||||||
|
# from eopkg util module (https://github.com/solus-project/package-management/blob/master/pisi/util.py#L293)
|
||||||
|
def join_path(a, *p) -> str:
|
||||||
|
"""Join two or more pathname components.
|
||||||
|
Python os.path.join cannot handle '/' at the start of latter components.
|
||||||
|
"""
|
||||||
|
for b in p:
|
||||||
|
b = b.lstrip('/')
|
||||||
|
if a == '' or a.endswith('/'):
|
||||||
|
a += b
|
||||||
|
continue
|
||||||
|
a += '/' + b
|
||||||
|
return a
|
||||||
|
|
||||||
|
|
||||||
|
def rack_bottle(fp: TextIOWrapper | str, f_name: str) -> bool:
|
||||||
|
"""Store a bottle file in the database and update package cache."""
|
||||||
|
if isinstance(fp, TextIOWrapper):
|
||||||
|
_log.debug(f"{fp=}, ensuring file is read from beginning...", end="")
|
||||||
|
try:
|
||||||
|
# ensure we are at the start of the file before reading
|
||||||
|
fp.seek(0)
|
||||||
|
c = fp.read()
|
||||||
|
except IOError as err:
|
||||||
|
_log.writeLog("failed.")
|
||||||
|
_log.error(f"Couldn't open file for reading: {str(err)}")
|
||||||
|
return False
|
||||||
|
finally:
|
||||||
|
_log.writeLog("ok.")
|
||||||
|
fp.close()
|
||||||
|
|
||||||
|
f_name = fp.name.split('/')[-1]
|
||||||
|
else:
|
||||||
|
_log.debug("fp is a string. The contents can be directly used.")
|
||||||
|
c = fp
|
||||||
|
|
||||||
|
# clean the file of any comments (specific to storing from user input)
|
||||||
|
if re.search(re_desc_tag, c):
|
||||||
|
_log.debug("Scrubbing bottle comments from string content.")
|
||||||
|
c = re.sub(re_desc_tag, "", c)
|
||||||
|
|
||||||
|
result = ensure_dir(soypak.db.rack_db_dir)
|
||||||
|
if result[0] != 0:
|
||||||
|
_log.error(f"There was an issue when ensuring directory (exit code: {result[0]}): {result[1]}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
new_file = join_path(soypak.db.rack_db_dir, f_name)
|
||||||
|
_log.debug(f"Opening new file to write bottle file to permanently: {new_file!r}")
|
||||||
|
try:
|
||||||
|
fp = open(new_file, 'w')
|
||||||
|
except IOError as err:
|
||||||
|
_log.error(f"There was an issue attempting to write to {new_file!r}: {str(err)}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
fp.write(c)
|
||||||
|
s = fp.tell()
|
||||||
|
fp.close()
|
||||||
|
|
||||||
|
_log.debug("Successfully wrote %s bytes to file." % s)
|
||||||
|
return True
|
||||||
Loading…
Reference in New Issue
Block a user