-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsearch_query.py
More file actions
59 lines (47 loc) · 2.15 KB
/
search_query.py
File metadata and controls
59 lines (47 loc) · 2.15 KB
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
from dataclasses import dataclass
from backupchan_server import models
@dataclass
class SearchQuery:
name: str | None
target_type: models.BackupType | None
recycle_criteria: models.BackupRecycleCriteria | None
recycle_action: models.BackupRecycleAction | None
location: str | None
name_template: str | None
deduplicate: bool | None
alias: str | None
tags: list[str] | None
def sql(self) -> tuple[str, list[any]]:
conditions = []
values = []
if self.name:
conditions.append("name LIKE %s")
values.append(f"%{self.name}%")
if self.target_type:
conditions.append("type = %s")
values.append(self.target_type)
if self.recycle_criteria:
conditions.append("recycle_criteria = %s")
values.append(self.recycle_criteria)
if self.recycle_action:
conditions.append("recycle_action = %s")
values.append(self.recycle_action)
if self.location:
conditions.append("location LIKE %s")
values.append(f"%{self.location}%")
if self.name_template:
conditions.append("name_template LIKE %s")
values.append(f"%{self.name_template}%")
if self.deduplicate is not None:
conditions.append("deduplicate = %s")
values.append(int(self.deduplicate))
if self.alias:
conditions.append("alias LIKE %s")
values.append(f"%{self.alias}%")
tags = [] if not self.tags else list(set(tag.strip() for tag in self.tags if tag.strip()))
if tags:
placeholders = ",".join(["%s"] * len(tags))
filter_string = ("AND " + " AND ".join(["tar." + condition for condition in conditions])) if conditions else ""
return f"SELECT tar.* FROM targets tar JOIN target_tags tt ON tar.id = tt.target_id JOIN tags t ON t.id = tt.tag_id WHERE t.name IN ({placeholders}) {filter_string} GROUP BY tar.id HAVING COUNT(DISTINCT t.name) >= %s", tags + values + [len(tags)]
filter_string = " AND ".join(conditions)
return f"SELECT * FROM targets WHERE {filter_string}", values