opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-08-31 01:02:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix new UP031 errors from ruff 0.4.2

Browse Source
This commit is contained in:
Nicola Soranzo
2024-04-27 16:03:22 +01:00
parent 0e89124a13
commit 0629908541
127 changed files with 381 additions and 425 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/api/common.py
+5 -5
View File
@@ -36,7 +36,7 @@ def get(api_key, url):
try:
return json.loads(urlopen(url).read())
except ValueError as e:
print("URL did not return JSON data: %s" % e)
print(f"URL did not return JSON data: {e}")
sys.exit(1)
@@ -87,14 +87,14 @@ def display(api_key, url, return_formatted=True):
print("------------------")
for n, i in enumerate(r):
if isinstance(i, str):
print(" %s" % i)
print(f" {i}")
else:
# All collection members should have a name in the response.
# url is optional
if "url" in i:
print("#%d: %s" % (n + 1, i.pop("url")))
if "name" in i:
print(" name: %s" % i.pop("name"))
print(f" name: {i.pop('name')}")
try:
for k, v in i.items():
print(f" {k}: {v}")
@@ -112,7 +112,7 @@ def display(api_key, url, return_formatted=True):
elif isinstance(r, str):
print(r)
else:
print("response is unknown type: %s" % type(r))
print(f"response is unknown type: {type(r)}")
def submit(api_key, url, data, return_formatted=True):
@@ -143,7 +143,7 @@ def submit(api_key, url, data, return_formatted=True):
else:
print("----")
if "name" in i:
print(" name: %s" % i.pop("name"))
print(f" name: {i.pop('name')}")
for k, v in i.items():
print(f" {k}: {v}")
else:
scripts/api/data_manager_example_execute.py
+2 -2
View File
@@ -47,7 +47,7 @@ def run_tool(tool_id, history_id, params, api_key, galaxy_url, wait=True, sleep_
def get_dataset_state(hda_id, api_key, galaxy_url):
datasets_url = urljoin(galaxy_url, "api/datasets/%s" % hda_id)
datasets_url = urljoin(galaxy_url, f"api/datasets/{hda_id}")
dataset_info = get(api_key, datasets_url)
return dataset_info["state"]
@@ -117,7 +117,7 @@ if __name__ == "__main__":
wait=False,
)
else:
"dbkey (%s) was specified more than once, skipping additional specification." % (dbkey)
f"dbkey ({dbkey}) was specified more than once, skipping additional specification."
print("Genomes Queued for downloading.")
scripts/api/display.py
+1 -1
View File
@@ -9,7 +9,7 @@ from common import display # noqa: I100,I202
try:
display(*sys.argv[1:3])
except TypeError as e:
print("usage: %s key url" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url")
print(e)
sys.exit(1)
except URLError as e:
scripts/api/example_watch_folder.py
+4 -4
View File
@@ -33,11 +33,11 @@ def main(api_key, api_url, in_folder, out_folder, data_library, workflow):
lib_create_data = {"name": data_library}
library = submit(api_key, api_url + "libraries", lib_create_data, return_formatted=False)
library_id = library[0]["id"]
folders = display(api_key, api_url + "libraries/%s/contents" % library_id, return_formatted=False)
folders = display(api_key, api_url + f"libraries/{library_id}/contents", return_formatted=False)
for f in folders:
if f["name"] == "/":
library_folder_id = f["id"]
workflow = display(api_key, api_url + "workflows/%s" % workflow, return_formatted=False)
workflow = display(api_key, api_url + f"workflows/{workflow}", return_formatted=False)
if not workflow:
print("Workflow %s not found, terminating.")
sys.exit(1)
@@ -57,7 +57,7 @@ def main(api_key, api_url, in_folder, out_folder, data_library, workflow):
data["upload_option"] = "upload_paths"
data["filesystem_paths"] = fullpath
data["create_type"] = "file"
libset = submit(api_key, api_url + "libraries/%s/contents" % library_id, data, return_formatted=False)
libset = submit(api_key, api_url + f"libraries/{library_id}/contents", data, return_formatted=False)
# TODO Handle this better, but the datatype isn't always
# set for the followup workflow execution without this
# pause.
@@ -88,6 +88,6 @@ if __name__ == "__main__":
data_library = sys.argv[5]
workflow = sys.argv[6]
except IndexError:
print("usage: %s key url in_folder out_folder data_library workflow" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url in_folder out_folder data_library workflow")
sys.exit(1)
main(api_key, api_url, in_folder, out_folder, data_library, workflow)
scripts/api/filter_failed_datasets_from_collection.py
+1 -2
View File
@@ -16,8 +16,7 @@ from bioblend.galaxy import (
if len(sys.argv) < 5:
print(
"Usage: %s <GalaxyUrl> <ApiKey> <HistoryName (must be unique)> <CollectionHistoryId (i.e. the simple integer id)>"
% sys.argv[0]
f"Usage: {sys.argv[0]} <GalaxyUrl> <ApiKey> <HistoryName (must be unique)> <CollectionHistoryId (i.e. the simple integer id)>"
)
exit(0)
scripts/api/history_create_history.py
+1 -1
View File
@@ -8,7 +8,7 @@ from common import submit
try:
assert sys.argv[2]
except IndexError:
print("usage: %s key url [name] " % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url [name] ")
sys.exit(1)
try:
data = {}
scripts/api/history_delete_history.py
+1 -1
View File
@@ -8,7 +8,7 @@ from common import delete
try:
assert sys.argv[2]
except IndexError:
print("usage: %s key url [purge (true/false)] " % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url [purge (true/false)] ")
sys.exit(1)
try:
data = {}
scripts/api/import_library_dataset_to_history.py
+1 -1
View File
@@ -10,7 +10,7 @@ try:
data = {}
data["from_ld_id"] = sys.argv[3]
except IndexError:
print("usage: %s key url library_file_id" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url library_file_id")
print(" library_file_id is from /api/libraries/<library_id>/contents/<library_file_id>")
sys.exit(1)
scripts/api/library_create_folder.py
+1 -1
View File
@@ -11,7 +11,7 @@ try:
data["name"] = sys.argv[4]
data["create_type"] = "folder"
except IndexError:
print("usage: %s key url folder_id name [description]" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url folder_id name [description]")
sys.exit(1)
try:
data["description"] = sys.argv[5]
scripts/api/library_create_library.py
+1 -1
View File
@@ -9,7 +9,7 @@ try:
data = {}
data["name"] = sys.argv[3]
except IndexError:
print("usage: %s key url name [description] [synopsys]" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url name [description] [synopsys]")
sys.exit(1)
try:
data["description"] = sys.argv[4]
scripts/api/library_upload_from_import_dir.py
+1 -1
View File
@@ -18,7 +18,7 @@ try:
data["upload_option"] = "upload_directory"
data["create_type"] = "file"
except IndexError:
print("usage: %s key url folder_id file_type server_dir dbkey" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url folder_id file_type server_dir dbkey")
sys.exit(1)
submit(sys.argv[1], sys.argv[2], data)
scripts/api/load_data_with_metadata.py
+2 -2
View File
@@ -39,7 +39,7 @@ def load_file(fullpath, api_key, api_url, library_id, library_folder_id, uuid_fi
if uuid_field is not None and uuid_field in ext_meta:
data["uuid"] = ext_meta[uuid_field]
libset = submit(api_key, api_url + "libraries/%s/contents" % library_id, data, return_formatted=True)
libset = submit(api_key, api_url + f"libraries/{library_id}/contents", data, return_formatted=True)
print(libset)
@@ -54,7 +54,7 @@ def main(api_key, api_url, in_folder, data_library, uuid_field=None):
lib_create_data = {"name": data_library}
library = submit(api_key, api_url + "libraries", lib_create_data, return_formatted=False)
library_id = library["id"]
folders = display(api_key, api_url + "libraries/%s/contents" % library_id, return_formatted=False)
folders = display(api_key, api_url + f"libraries/{library_id}/contents", return_formatted=False)
for f in folders:
if f["name"] == "/":
library_folder_id = f["id"]
scripts/api/reset_metadata_on_installed_repositories.py
+1 -1
View File
@@ -15,7 +15,7 @@ from common import submit
def main(options):
base_galaxy_url = options.galaxy_url.rstrip("/")
url = "%s/api/tool_shed_repositories/reset_metadata_on_installed_repositories" % base_galaxy_url
url = f"{base_galaxy_url}/api/tool_shed_repositories/reset_metadata_on_installed_repositories"
submit(options.api, url, {})
scripts/api/workflow_delete.py
+1 -1
View File
@@ -17,7 +17,7 @@ from common import delete
try:
assert sys.argv[2]
except IndexError:
print("usage: %s key url [purge (true/false)] " % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url [purge (true/false)] ")
sys.exit(1)
try:
data = {}
scripts/api/workflow_execute.py
+1 -1
View File
@@ -24,7 +24,7 @@ def main():
step, src, ds_id = v.split("=")
data["ds_map"][step] = {"src": src, "id": ds_id}
except IndexError:
print("usage: %s key url workflow_id history step=src=dataset_id" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url workflow_id history step=src=dataset_id")
sys.exit(1)
submit(sys.argv[1], sys.argv[2], data)
scripts/api/workflow_execute_parameters.py
+1 -1
View File
@@ -46,7 +46,7 @@ def main():
print("TOOL ID ERROR:")
except IndexError:
print("usage: %s key url workflow_id history step=src=dataset_id" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url workflow_id history step=src=dataset_id")
sys.exit(1)
submit(sys.argv[1], sys.argv[2], data)
scripts/api/workflow_import.py
+2 -2
View File
@@ -14,14 +14,14 @@ from common import submit
def main():
api_key = sys.argv[1]
api_base_url = sys.argv[2]
api_url = "%s/api/workflows" % api_base_url
api_url = f"{api_base_url}/api/workflows"
try:
data = {}
data["installed_repository_file"] = sys.argv[3]
if len(sys.argv) > 4 and sys.argv[4] == "--add_to_menu":
data["add_to_menu"] = True
except IndexError:
print("usage: %s key galaxy_url workflow_file" % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key galaxy_url workflow_file")
sys.exit(1)
submit(api_key, api_url, data, return_formatted=False)
scripts/api/workflow_import_from_file_rpark.py
+1 -1
View File
@@ -20,7 +20,7 @@ def openWorkflow(in_file):
try:
assert sys.argv[2]
except IndexError:
print("usage: %s key url [name] " % os.path.basename(sys.argv[0]))
print(f"usage: {os.path.basename(sys.argv[0])} key url [name] ")
sys.exit(1)
try:
data = {}
scripts/apply_tags.py
+3 -3
View File
@@ -29,12 +29,12 @@ class ApplyTagsHistory:
try:
update_history = history.show_history(self.history_id)
except Exception as exception:
print("Some problem occurred with history: %s" % self.history_id)
print(f"Some problem occurred with history: {self.history_id}")
print(exception)
return
update_history_id = update_history["id"]
print("History name: %s" % update_history["name"])
print("History id: %s" % update_history_id)
print(f"History name: {update_history['name']}")
print(f"History id: {update_history_id}")
self.find_dataset_parents_update_tags(history, job, update_history_id)
def find_dataset_parents_update_tags(self, history, job, history_id):
scripts/build_toolbox.py
+1 -1
View File
@@ -141,7 +141,7 @@ def scanfiles(filenamelist):
newtoolelement = ET.Element("tool", attrib)
toolboxpositionelements = toolelement.findall("toolboxposition")
if not toolboxpositionelements:
print("DBG> %s has no toolboxposition" % fn)
print(f"DBG> {fn} has no toolboxposition")
else:
for toolboxpositionelement in toolboxpositionelements:
toolbox.add(newtoolelement, toolboxpositionelement)
scripts/check_python.py
+4 -3
View File
@@ -3,6 +3,8 @@ If the current installed Python version is not supported, prints an error
message to stderr and returns 1
"""
from __future__ import print_function # noqa: UP010
import sys
@@ -13,13 +15,12 @@ def check_python():
else:
version_string = ".".join(str(_) for _ in sys.version_info[:3])
msg = (
"""\
ERROR: Your Python version is: %s
"""ERROR: Your Python version is: %s
Galaxy is currently supported on Python >=3.8 .
To run Galaxy, please install a supported Python version.
If a supported version is already installed but is not your default,
https://docs.galaxyproject.org/en/latest/admin/python.html contains instructions
on how to force Galaxy to use a different version."""
on how to force Galaxy to use a different version.""" # noqa: UP031
% version_string
)
print(msg, file=sys.stderr)
scripts/cleanup_datasets/admin_cleanup_datasets.py
+5 -5
View File
@@ -151,7 +151,7 @@ def main():
template_file = args.template
if template_file is None:
default_template = os.path.join(scriptdir, "admin_cleanup_deletion_template.txt")
sample_template_file = "%s.sample" % default_template
sample_template_file = f"{default_template}.sample"
if os.path.exists(default_template):
template_file = default_template
elif os.path.exists(sample_template_file):
@@ -164,7 +164,7 @@ def main():
"found, please specify template as an option (--template)."
)
elif not os.path.exists(template_file):
parser.error("Specified template file (%s) not found." % template_file)
parser.error(f"Specified template file ({template_file}) not found.")
config = galaxy.config.Configuration(**app_properties)
@@ -273,9 +273,9 @@ def administrative_delete_datasets(
subject = "Galaxy Server Cleanup " "- %d datasets DELETED" % len(dataset_list)
fromaddr = config.email_from
print()
print("From: %s" % fromaddr)
print("To: %s" % email)
print("Subject: %s" % subject)
print(f"From: {fromaddr}")
print(f"To: {email}")
print(f"Subject: {subject}")
print("----------")
print(msgtext)
if not info_only:
scripts/cleanup_datasets/pgcleanup.py
+9 -9
View File
@@ -143,22 +143,22 @@ class Action:
else:
logf = os.path.join(self._log_dir, self.name + ".log")
if self._dry_run:
log.info("--dry-run specified, logging changes to stderr instead of log file: %s" % logf)
log.info("--dry-run specified, logging changes to stderr instead of log file: %s", logf)
h = set_log_handler()
else:
log.info("Opening log file: %s" % logf)
log.info("Opening log file: %s", logf)
h = set_log_handler(filename=logf)
h.setLevel(logging.DEBUG if self._debug else logging.INFO)
h.setFormatter(LevelFormatter())
self.__log = logging.getLogger(self.name)
self.__log.addHandler(h)
self.__log.propagate = False
m = ("==== Log opened: %s " % datetime.datetime.now().isoformat()).ljust(72, "=")
m = (f"==== Log opened: {datetime.datetime.now().isoformat()} ").ljust(72, "=")
self.__log.info(m)
self.__log.info(f"Epoch time for this action: {self._epoch_time}")
self.__log.info("Epoch time for this action: %s", self._epoch_time)
def __close_log(self):
m = ("==== Log closed: %s " % datetime.datetime.now().isoformat()).ljust(72, "=")
m = (f"==== Log closed: {datetime.datetime.now().isoformat()} ").ljust(72, "=")
self.log.info(m)
self.__log = None
@@ -1221,7 +1221,7 @@ class Cleanup:
self.__conn = psycopg2.connect(cursor_factory=NamedTupleCursor, **args)
# TODO: is this per session or cursor?
if self.args.work_mem is not None:
log.info("Setting work_mem to %s" % self.args.work_mem)
log.info("Setting work_mem to %s", self.args.work_mem)
self.__conn.cursor().execute("SET work_mem TO %s", (self.args.work_mem,))
return self.__conn
@@ -1271,7 +1271,7 @@ class Cleanup:
nargs="*",
metavar="ACTION",
default=[],
help="Action(s) to perform, chosen from: %s" % ", ".join(sorted(self.actions.keys())),
help="Action(s) to perform, chosen from: {}".format(", ".join(sorted(self.actions.keys()))),
)
self.args = parser.parse_args()
@@ -1292,7 +1292,7 @@ class Cleanup:
ok = True
for name in self.args.actions:
if name not in self.actions.keys():
log.error("Unknown action in sequence: %s" % name)
log.error("Unknown action in sequence: %s", name)
ok = False
if not ok:
log.critical("Exiting due to previous error(s)")
@@ -1381,7 +1381,7 @@ class Cleanup:
self.__current_action = name
with cls(self) as action:
self._run_action(action)
log.info("Finished %s" % name)
log.info("Finished %s", name)
if __name__ == "__main__":
scripts/config_sample_to_kwalify.py
+4 -4
View File
@@ -55,14 +55,14 @@ def _dump_option(option, current_section_desc):
type = "str"
if default == "None":
default = None
print_line("%s:" % key)
print_line(" type: %s" % type)
print_line(f"{key}:")
print_line(f" type: {type}")
if default is not None:
print_line(" default: %s" % default)
print_line(f" default: {default}")
# print_line(" required: false")
print_line(" desc: |")
for line in current_section_desc:
print_line(" %s" % line)
print_line(f" {line}")
print_line("")
scripts/db_shell.py
+3 -3
View File
@@ -85,7 +85,7 @@ def printquery(statement, bind=None):
"""
if isinstance(value, str):
value = value.replace("'", "''")
return "'%s'" % value
return f"'{value}'"
elif value is None:
return "NULL"
elif isinstance(value, (float, int)):
@@ -93,10 +93,10 @@ def printquery(statement, bind=None):
elif isinstance(value, decimal.Decimal):
return str(value)
elif isinstance(value, datetime.datetime):
return "TO_DATE('%s','YYYY-MM-DD HH24:MI:SS')" % value.strftime("%Y-%m-%d %H:%M:%S")
return f"TO_DATE('{value.strftime('%Y-%m-%d %H:%M:%S')}','YYYY-MM-DD HH24:MI:SS')"
else:
raise NotImplementedError("Don't know how to literal-quote value %r" % value)
raise NotImplementedError(f"Don't know how to literal-quote value {value!r}")
compiler = LiteralCompiler(dialect, statement)
print(compiler.process(statement))
scripts/drmaa_external_killer.py
+1 -1
View File
@@ -14,7 +14,7 @@ import drmaa
def validate_paramters():
if len(sys.argv) < 3:
sys.stderr.write("usage: %s [job ID] [user uid]\n" % sys.argv[0])
sys.stderr.write(f"usage: {sys.argv[0]} [job ID] [user uid]\n")
exit(1)
jobID = sys.argv[1]
scripts/drmaa_external_runner.py
+3 -3
View File
@@ -50,14 +50,14 @@ def get_user_id_by_name(username):
try:
pw = pwd.getpwnam(username)
except KeyError:
sys.stderr.write("error: User name (%s) is not valid.\n" % username)
sys.stderr.write(f"error: User name ({username}) is not valid.\n")
exit(1)
return pw.pw_uid
def json_file_exists(json_filename):
if not os.path.exists(json_filename):
sys.stderr.write("error: JobTemplate file (%s) doesn't exist\n" % (json_filename))
sys.stderr.write(f"error: JobTemplate file ({json_filename}) doesn't exist\n")
exit(1)
return True
@@ -70,7 +70,7 @@ def validate_paramters():
sys.argv.remove("--assign_all_groups")
if len(sys.argv) < 3:
sys.stderr.write("usage: %s [USER-ID] [JSON-JOB-TEMPLATE-FILE]\n" % sys.argv[0])
sys.stderr.write(f"usage: {sys.argv[0]} [USER-ID] [JSON-JOB-TEMPLATE-FILE]\n")
exit(1)
userid = sys.argv[1]
scripts/dump_library_paths.py
+2 -2
View File
@@ -138,11 +138,11 @@ def _get_library_dataset_paths(args, kwargs):
filename = object_store.get_filename(dataset)
files_dir = dataset.get_extra_files_path()
if (args.exists and object_store.exists(dataset)) or not args.exists:
output.write("%s\n" % _path(filename, args))
output.write(f"{_path(filename, args)}\n")
elif args.exists:
log.warning("Missing %s", filename)
if files_dir and os.path.exists(files_dir):
output.write("%s\n" % _path(files_dir, args))
output.write(f"{_path(files_dir, args)}\n")
last_library = library
output.close()
scripts/external_chown_script.py
+3 -3
View File
@@ -15,7 +15,7 @@ ALLOWED_PATHS = None
def validate_parameters():
if len(sys.argv) < 4:
sys.stderr.write("usage: %s path user_name gid\n" % sys.argv[0])
sys.stderr.write(f"usage: {sys.argv[0]} path user_name gid\n")
exit(1)
path = os.path.abspath(sys.argv[1])
@@ -28,7 +28,7 @@ def validate_parameters():
allowed = True
break
if not allowed:
sys.stderr.write("owner and group modifications in %s are not allowed\n" % path)
sys.stderr.write(f"owner and group modifications in {path} are not allowed\n")
sys.exit(1)
galaxy_user_name = sys.argv[2]
@@ -44,7 +44,7 @@ def main():
(stdoutdata, stderrdata) = p.communicate()
exitcode = p.returncode
if exitcode != 0:
sys.exit("external_chown_script: could not chown\ncmd was %s\n" % " ".join(cmd))
sys.exit("external_chown_script: could not chown\ncmd was {}\n".format(" ".join(cmd)))
if __name__ == "__main__":
scripts/grt/export.py
+1 -1
View File
@@ -56,7 +56,7 @@ def kw_metrics(job):
def round_to_2sd(number):
if number:
return str(int(float("%.2g" % number)))
return str(int(float(f"{number:.2g}")))
else:
return "-1"
scripts/loc_files/create_all_fasta_loc.py
+27 -34
View File
@@ -272,16 +272,16 @@ def __main__():
paths_to_look_in = [os.path.join(options.genome_dir, "%s")]
# say what we're looking in
print("\nLooking in:\n\t%s" % "\n\t".join(p % "<build_name>" for p in paths_to_look_in))
poss_names = ["<build_name>%s" % _ for _ in variants]
print("for files that are named %s" % ", ".join(poss_names[:-1]), end=" ")
print("\nLooking in:\n\t{}".format("\n\t".join(p % "<build_name>" for p in paths_to_look_in)))
poss_names = [f"<build_name>{_}" for _ in variants]
print("for files that are named {}".format(", ".join(poss_names[:-1])), end=" ")
if len(poss_names) > 1:
print("or %s" % poss_names[-1], end=" ")
print(f"or {poss_names[-1]}", end=" ")
if len(options.fasta_exts) == 1:
print("with the extension %s." % ", ".join(fasta_exts[:-1]))
print("with the extension {}.".format(", ".join(fasta_exts[:-1])))
else:
print("with the extension {} or {}.".format(", ".join(fasta_exts[:-1]), fasta_exts[-1]))
print("\nSkipping the following:\n\t%s" % "\n\t".join(exemptions))
print("\nSkipping the following:\n\t{}".format("\n\t".join(exemptions)))
# get column names
col_values = []
@@ -350,40 +350,33 @@ def __main__():
# output results
print(
"\nThere were %s fasta files found that were not included because they did not have the expected file names."
% len(unmatching_fasta_paths)
f"\nThere were {len(unmatching_fasta_paths)} fasta files found that were not included because they did not have the expected file names."
)
print("%s fasta files were found and listed.\n" % len(fasta_locs.keys()))
print(f"{len(fasta_locs.keys())} fasta files were found and listed.\n")
# output unmatching fasta files
if options.unmatching_fasta and unmatching_fasta_paths:
open(options.unmatching_fasta, "wb").write("%s\n" % "\n".join(unmatching_fasta_paths))
open(options.unmatching_fasta, "wb").write("{}\n".format("\n".join(unmatching_fasta_paths)))
# output loc file
if options.append:
all_fasta_loc = open(loc_path, "ab")
else:
all_fasta_loc = open(loc_path, "wb")
# put sample loc file text at top of file if appropriate
if sample_text:
if options.loc_sample_name:
all_fasta_loc.write("%s\n" % open(options.loc_sample_name, "rb").read().strip())
else:
all_fasta_loc.write("%s\n" % open("%s.sample" % loc_path, "rb").read().strip())
# output list of fasta files in alphabetical order
fasta_bases = list(fasta_locs.keys())
fasta_bases.sort(key=str.upper)
for fb in fasta_bases:
out_line = []
for col in col_values:
try:
out_line.append(fasta_locs[fb][col])
except KeyError:
raise Exception("Unexpected column (%s) encountered" % col)
if out_line:
all_fasta_loc.write("%s\n" % "\t".join(out_line))
# close up output loc file
all_fasta_loc.close()
with open(loc_path, "ab" if options.append else "wb") as all_fasta_loc:
# put sample loc file text at top of file if appropriate
if sample_text:
loc_sample_name = options.loc_sample_name if options.loc_sample_name else f"{loc_path}.sample"
with open(loc_sample_name, "rb") as loc_sample_name_fh:
all_fasta_loc.write(f"{loc_sample_name_fh.read().strip()}\n")
# output list of fasta files in alphabetical order
fasta_bases = list(fasta_locs.keys())
fasta_bases.sort(key=str.upper)
for fb in fasta_bases:
out_line = []
for col in col_values:
try:
out_line.append(fasta_locs[fb][col])
except KeyError:
raise Exception(f"Unexpected column ({col}) encountered")
if out_line:
all_fasta_loc.write("{}\n".format("\t".join(out_line)))
if __name__ == "__main__":
scripts/microbes/create_bacteria_loc_file.py
+7 -7
View File
@@ -95,11 +95,11 @@ def __main__():
chr["chromosome"],
"sequence",
"fasta",
os.path.join(org["base_dir"], "%s.fna" % chr["chromosome"]),
os.path.join(org["base_dir"], f"{chr['chromosome']}.fna"),
)
)
# GeneMark
if os.path.exists(os.path.join(org["base_dir"], "%s.GeneMark.bed" % chr["chromosome"])):
if os.path.exists(os.path.join(org["base_dir"], f"{chr['chromosome']}.GeneMark.bed")):
print(
"DATA\t{}_{}_{}\t{}\t{}\t{}\t{}\t{}".format(
build,
@@ -109,11 +109,11 @@ def __main__():
chr["chromosome"],
"GeneMark",
"bed",
os.path.join(org["base_dir"], "%s.GeneMark.bed" % chr["chromosome"]),
os.path.join(org["base_dir"], f"{chr['chromosome']}.GeneMark.bed"),
)
)
# GenMarkHMM
if os.path.exists(os.path.join(org["base_dir"], "%s.GeneMarkHMM.bed" % chr["chromosome"])):
if os.path.exists(os.path.join(org["base_dir"], f"{chr['chromosome']}.GeneMarkHMM.bed")):
print(
"DATA\t{}_{}_{}\t{}\t{}\t{}\t{}\t{}".format(
build,
@@ -123,11 +123,11 @@ def __main__():
chr["chromosome"],
"GeneMarkHMM",
"bed",
os.path.join(org["base_dir"], "%s.GeneMarkHMM.bed" % chr["chromosome"]),
os.path.join(org["base_dir"], f"{chr['chromosome']}.GeneMarkHMM.bed"),
)
)
# Glimmer3
if os.path.exists(os.path.join(org["base_dir"], "%s.Glimmer3.bed" % chr["chromosome"])):
if os.path.exists(os.path.join(org["base_dir"], f"{chr['chromosome']}.Glimmer3.bed")):
print(
"DATA\t{}_{}_{}\t{}\t{}\t{}\t{}\t{}".format(
build,
@@ -137,7 +137,7 @@ def __main__():
chr["chromosome"],
"Glimmer3",
"bed",
os.path.join(org["base_dir"], "%s.Glimmer3.bed" % chr["chromosome"]),
os.path.join(org["base_dir"], f"{chr['chromosome']}.Glimmer3.bed"),
)
)
scripts/microbes/create_nib_seq_loc_file.py
+2 -2
View File
@@ -74,8 +74,8 @@ def __main__():
for chr in org["chrs"]:
chr = org["chrs"][chr]
fasta_file = os.path.join(org["base_dir"], "%s.fna" % chr["chromosome"])
nib_out_file = os.path.join(seq_path, "%s.nib " % chr["chromosome"])
fasta_file = os.path.join(org["base_dir"], f"{chr['chromosome']}.fna")
nib_out_file = os.path.join(seq_path, f"{chr['chromosome']}.nib ")
# create nibs using faToNib binary
# TODO: when bx supports writing nib, use it here instead
command = f"faToNib {fasta_file} {nib_out_file}"
scripts/microbes/harvest_bacteria.py
+18 -18
View File
@@ -158,18 +158,18 @@ def process_FASTA(filename, org_num, refseq):
fasta = "".join(fasta)
# Create Chrom Info File:
chrom_info_file = open(os.path.join(os.path.split(filename)[0], "%s.info" % refseq), "wb+")
chrom_info_file = open(os.path.join(os.path.split(filename)[0], f"{refseq}.info"), "wb+")
chrom_info_file.write(f"chromosome={refseq}\nname={chr_name}\nlength={len(fasta)}\norganism={org_num}\n")
try:
chrom_info_file.write("gi=%s\n" % accesions["gi"])
chrom_info_file.write(f"gi={accesions['gi']}\n")
except Exception:
chrom_info_file.write("gi=None\n")
try:
chrom_info_file.write("gb=%s\n" % accesions["gb"])
chrom_info_file.write(f"gb={accesions['gb']}\n")
except Exception:
chrom_info_file.write("gb=None\n")
try:
chrom_info_file.write("refseq=%s\n" % refseq)
chrom_info_file.write(f"refseq={refseq}\n")
except Exception:
chrom_info_file.write("refseq=None\n")
chrom_info_file.close()
@@ -191,7 +191,7 @@ def process_Glimmer3(filename, org_num, refseq):
except Exception as e:
print("Converting Glimmer3 to bed FAILED! For chrom:", refseq, "file:", filename, e)
glimmer3_bed = []
glimmer3_bed_file = open(os.path.join(os.path.split(filename)[0], "%s.Glimmer3.bed" % refseq), "wb+")
glimmer3_bed_file = open(os.path.join(os.path.split(filename)[0], f"{refseq}.Glimmer3.bed"), "wb+")
glimmer3_bed_file.write("\n".join(glimmer3_bed))
glimmer3_bed_file.close()
@@ -202,7 +202,7 @@ def process_GeneMarkHMM(filename, org_num, refseq):
except Exception as e:
print("Converting GeneMarkHMM to bed FAILED! For chrom:", refseq, "file:", filename, e)
geneMarkHMM_bed = []
geneMarkHMM_bed_bed_file = open(os.path.join(os.path.split(filename)[0], "%s.GeneMarkHMM.bed" % refseq), "wb+")
geneMarkHMM_bed_bed_file = open(os.path.join(os.path.split(filename)[0], f"{refseq}.GeneMarkHMM.bed"), "wb+")
geneMarkHMM_bed_bed_file.write("\n".join(geneMarkHMM_bed))
geneMarkHMM_bed_bed_file.close()
@@ -213,7 +213,7 @@ def process_GeneMark(filename, org_num, refseq):
except Exception as e:
print("Converting GeneMark to bed FAILED! For chrom:", refseq, "file:", filename, e)
geneMark_bed = []
geneMark_bed_bed_file = open(os.path.join(os.path.split(filename)[0], "%s.GeneMark.bed" % refseq), "wb+")
geneMark_bed_bed_file = open(os.path.join(os.path.split(filename)[0], f"{refseq}.GeneMark.bed"), "wb+")
geneMark_bed_bed_file.write("\n".join(geneMark_bed))
geneMark_bed_bed_file.close()
@@ -228,9 +228,9 @@ def __main__():
try:
os.mkdir(base_dir)
print("path '%s' has been created" % base_dir)
print(f"path '{base_dir}' has been created")
except Exception:
print("path '%s' seems to already exist" % base_dir)
print(f"path '{base_dir}' seems to already exist")
for org_num, name, chroms, kingdom, group, _, _, info_url, ftp_url in iter_genome_projects():
if chroms is None:
@@ -240,7 +240,7 @@ def __main__():
org_dir = os.path.join(base_dir, org_num)
os.mkdir(org_dir)
except Exception:
print("Organism %s already exists on disk, skipping" % org_num)
print(f"Organism {org_num} already exists on disk, skipping")
continue
# get ftp contents
@@ -254,14 +254,14 @@ def __main__():
print("Org:", org_num, "chrom:", refseq, "[", time.time() - start_time, "seconds elapsed. ]")
# Create org info file
info_file = open(os.path.join(org_dir, "%s.info" % org_num), "wb+")
info_file.write("genome project id=%s\n" % org_num)
info_file.write("name=%s\n" % name)
info_file.write("kingdom=%s\n" % kingdom)
info_file.write("group=%s\n" % group)
info_file.write("chromosomes=%s\n" % ",".join(chroms))
info_file.write("info url=%s\n" % info_url)
info_file.write("ftp url=%s\n" % ftp_url)
info_file = open(os.path.join(org_dir, f"{org_num}.info"), "wb+")
info_file.write(f"genome project id={org_num}\n")
info_file.write(f"name={name}\n")
info_file.write(f"kingdom={kingdom}\n")
info_file.write(f"group={group}\n")
info_file.write("chromosomes={}\n".format(",".join(chroms)))
info_file.write(f"info url={info_url}\n")
info_file.write(f"ftp url={ftp_url}\n")
info_file.close()
print("Finished Harvesting", "[", time.time() - start_time, "seconds elapsed. ]")
scripts/pages_identifier_conversion.py
+2 -2
View File
@@ -58,11 +58,11 @@ def main(argv):
with transaction(session):
session.commit()
else:
print("Modifying revision %s." % p.id)
print(f"Modifying revision {p.id}.")
print(difflib.unified_diff(p.content, newcontent))
except Exception:
logging.exception(
"Error parsing page, rolling changes back and skipping revision %s. Please report this error." % p.id
"Error parsing page, rolling changes back and skipping revision %s. Please report this error.", p.id
)
session.rollback()
scripts/runtime_stats.py
+12 -12
View File
@@ -90,7 +90,7 @@ def parse_arguments():
parser.add_argument("-M", "--max", type=int, default=-1, help="Ignore runtimes greater than MAX seconds")
parser.add_argument("-u", "--user", help="Return stats for only this user (id, email, " "or username)")
parser.add_argument(
"-s", "--source", default="metrics", help="Runtime data source (SOURCES: %s)" % ", ".join(DATA_SOURCES)
"-s", "--source", default="metrics", help="Runtime data source (SOURCES: {})".format(", ".join(DATA_SOURCES))
)
args = parser.parse_args()
@@ -144,13 +144,13 @@ def query(
if row:
user_id = row[0]
else:
print("Invalid user: %s" % user)
print(f"Invalid user: {user}")
sys.exit(1)
if like:
query_tool_id = "%%/%s/%%" % tool_id
query_tool_id = f"%/{tool_id}/%"
elif "/" in tool_id and not re.match(r"\d+\.\d+", tool_id.split("/")[-1]):
query_tool_id = "%s%%" % tool_id
query_tool_id = f"{tool_id}%"
like = True
else:
query_tool_id = tool_id
@@ -187,14 +187,14 @@ def query(
if min > 0 and max > 0:
time_clause = """WHERE ctimes[1] - ctimes[2] > interval %s
AND ctimes[1] - ctimes[2] < interval %s"""
sql_args.append("%s seconds" % min)
sql_args.append("%s seconds" % max)
sql_args.append(f"{min} seconds")
sql_args.append(f"{max} seconds")
elif min > 0:
time_clause = "WHERE ctimes[1] - ctimes[2] > interval %s"
sql_args.append("%s seconds" % min)
sql_args.append(f"{min} seconds")
elif max > 0:
time_clause = "WHERE ctimes[1] - ctimes[2] < interval %s"
sql_args.append("%s seconds" % max)
sql_args.append(f"{max} seconds")
else:
time_clause = ""
sql = HISTORY_SQL
@@ -218,7 +218,7 @@ def query(
return
if user:
print("Displaying statistics for user %s" % user)
print(f"Displaying statistics for user {user}")
stats = (
("Mean runtime", numpy.mean(times)),
@@ -229,11 +229,11 @@ def query(
for name, seconds in stats:
hours, minutes = nice_times(seconds)
msg = name + " is %0.0f seconds" % seconds
msg = name + f" is {seconds:0.0f} seconds"
if minutes:
msg += " (=%0.2f minutes)" % minutes
msg += f" (={minutes:0.2f} minutes)"
if hours:
msg += " (=%0.2f hours)" % hours
msg += f" (={hours:0.2f} hours)"
print(msg)
scripts/set_user_disk_usage.py
+2 -2
View File
@@ -62,9 +62,9 @@ def quotacheck(sa_session, users, engine, object_store):
print("none")
else:
if new > current:
print("+%s" % (nice_size(new - current)))
print(f"+{nice_size(new - current)}")
else:
print("-%s" % (nice_size(current - new)))
print(f"-{nice_size(current - new)}")
if __name__ == "__main__":
scripts/tests_markdown.py
+2 -2
View File
@@ -244,10 +244,10 @@ def __inject_api_timing_summary_test(test):
continue
endpoint_summary = {"total_time": summarize_times(timings), "label": endpoint[len("api.") :]}
sql_times = "sql.%s" % endpoint
sql_times = f"sql.{endpoint}"
if sql_times in timing:
endpoint_summary["sql_time"] = summarize_times(timing[sql_times])
sql_queries = "sqlqueries.%s" % endpoint
sql_queries = f"sqlqueries.{endpoint}"
if sql_queries in counter:
endpoint_summary["sql_queries"] = summarize_counter(counter[sql_queries])
api_endpoints[endpoint] = endpoint_summary
scripts/tool_shed/api/common.py
+3 -3
View File
@@ -35,7 +35,7 @@ def display(url, api_key=None, return_formatted=True):
if "url" in i:
print("#%d: %s" % (n + 1, i.pop("url")))
if "name" in i:
print(" name: %s" % i.pop("name"))
print(f" name: {i.pop('name')}")
for k, v in i.items():
print(f" {k}: {v}")
print()
@@ -47,7 +47,7 @@ def display(url, api_key=None, return_formatted=True):
for k, v in r.items():
print(f"{k}: {v}")
else:
print("response is unknown type: %s" % type(r))
print(f"response is unknown type: {type(r)}")
def get(url, api_key=None):
@@ -100,7 +100,7 @@ def submit(url, data, api_key=None, return_formatted=True):
else:
print("----")
if "name" in i:
print(" name: %s" % i.pop("name"))
print(f" name: {i.pop('name')}")
for k, v in i.items():
print(f" {k}: {v}")
else:
scripts/tool_shed/api/create_categories.py
+2 -2
View File
@@ -31,7 +31,7 @@ def main(options):
from_tool_shed = options.from_tool_shed.rstrip("/")
to_tool_shed = options.to_tool_shed.rstrip("/")
# Get the categories from the specified Tool Shed.
url = "%s/api/categories" % from_tool_shed
url = f"{from_tool_shed}/api/categories"
category_dicts = get(url)
create_response_dicts = []
for category_dict in category_dicts:
@@ -39,7 +39,7 @@ def main(options):
description = category_dict.get("description", None)
if name is not None and description is not None:
data = dict(name=name, description=description)
url = "%s/api/categories" % to_tool_shed
url = f"{to_tool_shed}/api/categories"
try:
response = submit(url, data, api_key)
except Exception as e:
scripts/tool_shed/api/create_users.py
+3 -3
View File
@@ -32,16 +32,16 @@ def main(options):
from_tool_shed = options.from_tool_shed.rstrip("/")
to_tool_shed = options.to_tool_shed.rstrip("/")
# Get the users from the specified Tool Shed.
url = "%s/api/users" % from_tool_shed
url = f"{from_tool_shed}/api/users"
user_dicts = get(url)
create_response_dicts = []
for user_dict in user_dicts:
username = user_dict.get("username", None)
if username is not None:
email = "%s@test.org" % username
email = f"{username}@test.org"
password = "testuser"
data = dict(email=email, password=password, username=username)
url = "%s/api/users" % to_tool_shed
url = f"{to_tool_shed}/api/users"
try:
response = submit(url, data, api_key)
except Exception as e:
scripts/tool_shed/deprecate_repositories_without_metadata.py
+4 -4
View File
@@ -68,7 +68,7 @@ def main():
try:
ini_file = args[0]
except IndexError:
sys.exit("Usage: python %s <tool shed .ini file> [options]" % sys.argv[0])
sys.exit(f"Usage: python {sys.argv[0]} <tool shed .ini file> [options]")
config_parser = configparser.ConfigParser({"here": os.getcwd()})
config_parser.read(ini_file)
config_dict = {}
@@ -103,7 +103,7 @@ def send_mail_to_owner(app, owner, email, repositories_deprecated, days=14):
elif url is None:
print("# Environment variable TOOL_SHED_CANONICAL_URL not set, not sending email to repository owner.")
return
subject = "Regarding your tool shed repositories at %s" % url
subject = f"Regarding your tool shed repositories at {url}"
message_body_template = (
"The tool shed automated repository checker has discovered that one or more of your repositories hosted "
+ "at this tool shed url ${url} have remained empty for over ${days} days, so they have been marked as deprecated. If you have plans "
@@ -123,7 +123,7 @@ def send_mail_to_owner(app, owner, email, repositories_deprecated, days=14):
)
return True
except Exception as e:
print("# An error occurred attempting to send email: %s" % e)
print(f"# An error occurred attempting to send email: {e}")
return False
@@ -193,7 +193,7 @@ class DeprecateRepositoriesApplication:
def __init__(self, config):
if config.database_connection is False:
config.database_connection = "sqlite:///%s?isolation_level=IMMEDIATE" % config.database
config.database_connection = f"sqlite:///{config.database}?isolation_level=IMMEDIATE"
# Setup the database engine and ORM
self.model = tool_shed.webapp.model.mapping.init(
config.file_path, config.database_connection, engine_options={}, create_tables=False
scripts/tools/re_escape_output.py
+1 -1
View File
@@ -27,7 +27,7 @@ def __main__():
if options.multiline:
lines = [re.escape(input.read())]
else:
lines = ["%s\n" % re.escape(line.rstrip("\n\r")) for line in input]
lines = ["{}\n".format(re.escape(line.rstrip("\n\r"))) for line in input]
output.writelines(lines)
output.close()
scripts/update_shed_config_path.py
+2 -2
View File
@@ -40,7 +40,7 @@ def create_database(config_file):
if database_connection is None:
database_connection = parser.get("app:main", "database_connection")
if database_connection is None:
database_connection = "sqlite:///%s" % parser.get("app:main", "database_file")
database_connection = "sqlite:///{}".format(parser.get("app:main", "database_file"))
if database_connection is None:
print("Unable to determine correct database connection.")
exit(1)
@@ -78,7 +78,7 @@ if __name__ == "__main__":
)
opts = parser.parse_args()
if not os.path.exists(opts.good_filename) and not opts.force:
print("The file %s does not exist, use the --force option to proceed." % opts.good_filename)
print(f"The file {opts.good_filename} does not exist, use the --force option to proceed.")
exit(1)
session, model = create_database(opts.config_file)
exit(main(opts, session, model))