1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
"""Delete phenotypes."""
import sys
import logging
from pathlib import Path
from typing import Optional
from urllib.parse import urljoin
from argparse import Namespace, ArgumentParser
import requests
from MySQLdb.cursors import DictCursor, BaseCursor
from gn_libs.mysqldb import database_connection
from uploader.phenotypes.models import delete_phenotypes
from scripts.cli.logging import setup_logging
from scripts.cli.options import (add_logging,
add_mariadb_uri,
add_population_id)
logger = logging.getLogger(__name__)
def read_xref_ids_file(filepath: Optional[Path]) -> tuple[int, ...]:
"""Read the phenotypes' cross-reference IDS from file."""
if filepath is None:
return tuple()
_ids: tuple[int, ...] = tuple()
with filepath.open(mode="r") as infile:
try:
_ids += (int(infile.readline().strip()),)
except TypeError:
pass
return _ids
def fetch_all_xref_ids(
cursor: BaseCursor, population_id: int) -> tuple[int, ...]:
"""Fetch all cross-reference IDs."""
cursor.execute("SELECT Id FROM PublishXRef WHERE InbredSetId=%s",
(population_id,))
return tuple(int(row["Id"]) for row in cursor.fetchall())
def update_auth(
auth_details: tuple[str, str],
species_id: int,
population_id: int,
dataset_id: int,
xref_ids: tuple[int, ...] = tuple()
):
"""Update the authorisation server: remove items to delete."""
authserver, token = auth_details
resp = requests.post(
urljoin(authserver,
(f"/auth/data/phenotypes/{species_id}/{population_id}"
f"/{dataset_id}/delete")),
timeout=(9.13, 20),
headers={
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
},
json={"xref_ids": xref_ids})
resp.raise_for_status()
def delete_the_phenotypes(
cursor: BaseCursor,
population_id: int,
xref_ids: tuple[int, ...] = tuple()) -> int:
"""Process and delete the phenotypes."""
delete_phenotypes(cursor, population_id, xref_ids)
return 0
if __name__ == "__main__":
def parse_args() -> Namespace:
"""Parse CLI arguments."""
parser = add_logging(
add_population_id(
add_mariadb_uri(
ArgumentParser(
prog="delete-phenotypes",
description=(
"Script to delete phenotypes from the database.")))))
parser.add_argument(
"dataset_id",
metavar="DATASET-ID",
type=int,
help="The dataset identifier for phenotypes to delete.")
parser.add_argument(
"auth_server_uri",
metavar="AUTH-SERVER-URI",
type=str,
help="URI to the authorisation server.")
parser.add_argument(
"auth_token",
metavar="AUTH-TOKEN",
type=str,
help=("Token to use to update the authorisation system with the "
"deletions done."))
parser.add_argument(
"--xref_ids_file",
metavar="XREF-IDS-FILE",
type=Path,
help=("Path to a file with phenotypes cross-reference IDs to "
"delete."))
parser.add_argument(
"--delete-all",
action="store_true",
help=("If no 'XREF-IDS-FILE' is provided, this flag determines "
"whether or not all the phenotypes for the given population "
"will be deleted."))
return parser.parse_args()
def main():
"""The `delete-phenotypes` script's entry point."""
args = parse_args()
setup_logging(logger, args.log_level.upper(), tuple())
with (database_connection(args.db_uri) as conn,
conn.cursor(cursorclass=DictCursor) as cursor):
xref_ids = read_xref_ids_file(args.xref_ids_file)
try:
assert not (len(xref_ids) > 0 and args.delete_all)
xref_ids = (fetch_all_xref_ids(cursor, args.population_id)
if args.delete_all else xref_ids)
if len(xref_ids) == 0:
print("No cross-reference IDs were provided. Aborting.")
return 0
update_auth((args.auth_server_uri, args.auth_token),
args.species_id,
args.population_id,
args.dataset_id,
xref_ids)
delete_phenotypes(cursor, args.population_id, xref_ids=xref_ids)
return 0
except AssertionError:
logger.error(
"'DELETE-ALL' and 'XREF-IDS' are mutually exclusive. "
"If you specify the list of XREF-IDS (in a file) to delete "
"and also specify to 'DELETE-ALL' phenotypes in the "
"population, we have no way of knowing what it is you want.")
return 1
except Exception as _exc:# pylint: disable=[broad-exception-caught]
logger.debug("Failed while attempting to delete phenotypes.",
exc_info=True)
return 1
sys.exit(main())
|
