mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Code cleanups from pyupgrade
Found by running: ``` ack --type=python -f | grep -v '^tools/' | xargs pyupgrade --py37-plus ```
This commit is contained in:
@@ -1382,7 +1382,7 @@ class ExtendedXYZ(XYZ):
|
||||
if properties is None: # re.search returned None
|
||||
raise ValueError(f"Could not find column properties in line: {comment}")
|
||||
properties = [s.split(":") for s in re.findall(r"[a-zA-Z]+:[SIRL]:[0-9]+", properties.group(1))]
|
||||
total_columns = sum([int(s[2]) for s in properties])
|
||||
total_columns = sum(int(s[2]) for s in properties)
|
||||
|
||||
for _ in range(n_atoms):
|
||||
atom_dict = {}
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import abc
|
||||
import os
|
||||
import time
|
||||
from typing import Set
|
||||
|
||||
from typing_extensions import ClassVar
|
||||
from typing import (
|
||||
ClassVar,
|
||||
Set,
|
||||
)
|
||||
|
||||
from galaxy.exceptions import (
|
||||
ConfigurationError,
|
||||
|
||||
@@ -4,6 +4,7 @@ import logging
|
||||
import os
|
||||
from typing import (
|
||||
Any,
|
||||
ClassVar,
|
||||
Dict,
|
||||
List,
|
||||
Optional,
|
||||
@@ -12,7 +13,6 @@ from typing import (
|
||||
|
||||
import fs
|
||||
from fs.base import FS
|
||||
from typing_extensions import ClassVar
|
||||
|
||||
from ..sources import BaseFilesSource
|
||||
|
||||
|
||||
@@ -36,7 +36,7 @@ def stream_url_to_str(
|
||||
) -> str:
|
||||
tmp_file = stream_url_to_file(path, file_sources=file_sources, prefix=prefix)
|
||||
try:
|
||||
with open(tmp_file, "r") as f:
|
||||
with open(tmp_file) as f:
|
||||
return f.read()
|
||||
finally:
|
||||
os.remove(tmp_file)
|
||||
|
||||
@@ -146,7 +146,7 @@ class WorkflowsManager(sharable.SharableModelManager):
|
||||
filters.append(model.StoredWorkflowUserShareAssociation.user == user)
|
||||
|
||||
if show_published or user is None and show_published is None:
|
||||
filters.append((model.StoredWorkflow.published == true()))
|
||||
filters.append(model.StoredWorkflow.published == true())
|
||||
|
||||
query = trans.sa_session.query(model.StoredWorkflow)
|
||||
if show_shared:
|
||||
|
||||
@@ -2257,7 +2257,7 @@ def source_to_import_store(
|
||||
delete = True
|
||||
target_path = source_uri
|
||||
if target_path.endswith(".json"):
|
||||
with open(target_path, "r") as f:
|
||||
with open(target_path) as f:
|
||||
store_dict = load(f)
|
||||
assert isinstance(store_dict, dict)
|
||||
model_import_store = get_import_model_store_for_dict(
|
||||
|
||||
@@ -50,7 +50,7 @@ def main(argv=None):
|
||||
|
||||
data: Dict[str, Any] = {}
|
||||
if is_json:
|
||||
with open(store_path, "r") as f:
|
||||
with open(store_path) as f:
|
||||
store_dict = yaml.safe_load(f)
|
||||
data["store_dict"] = store_dict
|
||||
else:
|
||||
|
||||
@@ -244,7 +244,7 @@ class ShortTermStorageManager(ShortTermStorageAllocator, ShortTermStorageMonitor
|
||||
|
||||
def _load_metadata(self, target_directory: Path, meta_name: str):
|
||||
meta_path = target_directory / f"{meta_name}.json"
|
||||
with open(meta_path, "r") as f:
|
||||
with open(meta_path) as f:
|
||||
return json.load(f)
|
||||
|
||||
def _directory(self, target: Union[str, ShortTermStorageTarget]) -> Path:
|
||||
|
||||
@@ -32,7 +32,7 @@ def _next_worker_id(server):
|
||||
if server._worker_id_overload:
|
||||
return server._worker_id_overload.pop()
|
||||
|
||||
in_use = set(w._worker_id for w in tuple(server.WORKERS.values()) if w.alive)
|
||||
in_use = {w._worker_id for w in tuple(server.WORKERS.values()) if w.alive}
|
||||
free = set(range(1, server._worker_id_current_workers + 1)) - in_use
|
||||
|
||||
return free.pop()
|
||||
|
||||
@@ -85,13 +85,13 @@ class JobFilesIntegerationTestCase(integration_util.IntegrationTestCase):
|
||||
post_url = self._api_url(f"jobs/{job_id}/files", use_key=False)
|
||||
response = requests.post(post_url, data=data, files=files())
|
||||
api_asserts.assert_status_code_is_ok(response)
|
||||
assert open(path, "r").read() == "some initial text data"
|
||||
assert open(path).read() == "some initial text data"
|
||||
|
||||
work_dir_file = os.path.join(working_directory, "work")
|
||||
data = {"path": work_dir_file, "job_key": job_key}
|
||||
response = requests.post(post_url, data=data, files=files())
|
||||
api_asserts.assert_status_code_is_ok(response)
|
||||
assert open(work_dir_file, "r").read() == "some initial text data"
|
||||
assert open(work_dir_file).read() == "some initial text data"
|
||||
|
||||
# set job state to finished and ensure the file is no longer
|
||||
# readable
|
||||
|
||||
@@ -337,6 +337,6 @@ def _assert_2_bed_metadata(hda: HistoryDatasetAssociation) -> None:
|
||||
|
||||
|
||||
def _assert_path_contains_2_bed(path) -> None:
|
||||
with open(path, "r") as f:
|
||||
with open(path) as f:
|
||||
contents = f.read()
|
||||
assert contents == CONTENTS_2_BED
|
||||
|
||||
Reference in New Issue
Block a user