Coverage for chalicelib/core/metadata.py: 43%
151 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1import re
2from typing import Optional
4from cachetools import TTLCache, cached
5from fastapi import HTTPException, status
7from chalicelib.core import projects
8from chalicelib.utils import pg_client
10MAX_INDEXES = 10
13def column_names():
14 return [f"metadata_{i}" for i in range(1, MAX_INDEXES + 1)]
17def __exists_by_name(project_id: int, name: str, exclude_index: Optional[int]) -> bool:
18 with pg_client.PostgresClient() as cur:
19 constraints = column_names()
20 if exclude_index:
21 del constraints[exclude_index - 1]
22 for i in range(len(constraints)):
23 constraints[i] += " ILIKE %(name)s"
24 query = cur.mogrify(f"""SELECT EXISTS(SELECT 1
25 FROM public.projects
26 WHERE project_id = %(project_id)s
27 AND deleted_at ISNULL
28 AND ({" OR ".join(constraints)})) AS exists;""",
29 {"project_id": project_id, "name": name})
30 cur.execute(query=query)
31 row = cur.fetchone()
33 return row["exists"]
36def get(project_id):
37 with pg_client.PostgresClient() as cur:
38 query = cur.mogrify(f"""SELECT {",".join(column_names())}
39 FROM public.projects
40 WHERE project_id = %(project_id)s
41 AND deleted_at ISNULL
42 LIMIT 1;""", {"project_id": project_id})
43 cur.execute(query=query)
44 metas = cur.fetchone()
45 results = []
46 if metas is not None: 46 ↛ 50line 46 didn't jump to line 50 because the condition on line 46 was always true
47 for i, k in enumerate(metas.keys()):
48 if metas[k] is not None:
49 results.append({"key": metas[k], "index": i + 1})
50 return results
53def get_batch(project_ids):
54 if project_ids is None or len(project_ids) == 0:
55 return []
56 with pg_client.PostgresClient() as cur:
57 query = cur.mogrify(f"""SELECT project_id, {",".join(column_names())}
58 FROM public.projects
59 WHERE project_id IN %(project_ids)s
60 AND deleted_at ISNULL;""",
61 {"project_ids": tuple(project_ids)})
62 cur.execute(query=query)
63 full_metas = cur.fetchall()
64 results = {}
65 if full_metas is not None and len(full_metas) > 0:
66 for metas in full_metas:
67 results[str(metas["project_id"])] = []
68 for i, k in enumerate(metas.keys()):
69 if metas[k] is not None and k != "project_id":
70 results[str(metas["project_id"])].append({"key": metas[k], "index": i + 1})
71 return results
74regex = re.compile(r'^[a-z0-9_-]+$', re.IGNORECASE)
77def index_to_colname(index):
78 if index <= 0 or index > MAX_INDEXES: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true
79 raise Exception("metadata index out or bound")
80 return f"metadata_{index}"
83def __get_available_index(project_id):
84 used_indexs = get(project_id)
85 used_indexs = [i["index"] for i in used_indexs]
86 if len(used_indexs) >= MAX_INDEXES: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true
87 return -1
88 i = 1
89 while i in used_indexs:
90 i += 1
91 return i
94def __edit(project_id, col_index, colname, new_name):
95 if new_name is None or len(new_name) == 0: 95 ↛ 97line 95 didn't jump to line 97 because the condition on line 95 was always true
96 return {"errors": ["key value invalid"]}
97 old_metas = get(project_id)
98 old_metas = {k["index"]: k for k in old_metas}
99 if col_index not in list(old_metas.keys()):
100 return {"errors": ["custom field not found"]}
102 if old_metas[col_index]["key"] != new_name:
103 with pg_client.PostgresClient() as cur:
104 query = cur.mogrify(f"""UPDATE public.projects
105 SET {colname} = %(value)s
106 WHERE project_id = %(project_id)s
107 AND deleted_at ISNULL
108 RETURNING {colname},
109 (SELECT {colname} FROM projects WHERE project_id = %(project_id)s) AS old_{colname};""",
110 {"project_id": project_id, "value": new_name})
111 cur.execute(query=query)
112 row = cur.fetchone()
113 new_name = row[colname]
114 old_name = row['old_' + colname]
115 old_metas[col_index]["key"] = new_name
116 projects.rename_metadata_condition(project_id=project_id,
117 old_metadata_key=old_name,
118 new_metadata_key=new_name)
119 return {"data": old_metas[col_index]}
122def edit(tenant_id, project_id, index: int, new_name: str):
123 if __exists_by_name(project_id=project_id, name=new_name, exclude_index=index): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true
124 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"name already exists.")
125 return __edit(project_id=project_id, col_index=index, colname=index_to_colname(index), new_name=new_name)
128def delete(tenant_id, project_id, index: int):
129 index = int(index)
130 old_segments = get(project_id)
131 old_indexes = [k["index"] for k in old_segments]
132 if index not in old_indexes: 132 ↛ 135line 132 didn't jump to line 135 because the condition on line 132 was always true
133 return {"errors": ["custom field not found"]}
135 with pg_client.PostgresClient() as cur:
136 colname = index_to_colname(index)
137 query = cur.mogrify(f"""UPDATE public.projects
138 SET {colname}= NULL
139 WHERE project_id = %(project_id)s AND deleted_at ISNULL;""",
140 {"project_id": project_id})
141 cur.execute(query=query)
142 projects.delete_metadata_condition(project_id=project_id,
143 metadata_key=old_segments[old_indexes.index(index)]["key"])
144 return {"data": get(project_id)}
147def add(tenant_id, project_id, new_name):
148 index = __get_available_index(project_id=project_id)
149 if index < 1: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true
150 return {"errors": ["maximum allowed metadata reached"]}
151 if __exists_by_name(project_id=project_id, name=new_name, exclude_index=None):
152 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"name already exists.")
153 with pg_client.PostgresClient() as cur:
154 colname = index_to_colname(index)
155 query = cur.mogrify(f"""UPDATE public.projects
156 SET {colname}= %(key)s
157 WHERE project_id =%(project_id)s
158 RETURNING {colname};""",
159 {"key": new_name, "project_id": project_id})
160 cur.execute(query=query)
161 col_val = cur.fetchone()[colname]
162 return {"data": {"key": col_val, "index": index}}
165def get_keys_by_projects(project_ids):
166 if project_ids is None or len(project_ids) == 0:
167 return {}
168 with pg_client.PostgresClient() as cur:
169 query = cur.mogrify(f"""SELECT project_id,{",".join(column_names())}
170 FROM public.projects
171 WHERE project_id IN %(project_ids)s
172 AND deleted_at ISNULL;""",
173 {"project_ids": tuple(project_ids)})
175 cur.execute(query)
176 rows = cur.fetchall()
177 results = {}
178 for r in rows:
179 project_id = r.pop("project_id")
180 results[project_id] = {}
181 for m in r:
182 if r[m] is not None:
183 results[project_id][m] = r[m]
184 return results
187def get_remaining_metadata_with_count(tenant_id):
188 all_projects = projects.get_projects(tenant_id=tenant_id)
189 results = []
190 used_metas = get_batch([p["projectId"] for p in all_projects])
191 for p in all_projects:
192 if MAX_INDEXES < 0:
193 remaining = -1
194 else:
195 remaining = MAX_INDEXES - len(used_metas[str(p["projectId"])])
196 results.append(
197 {**p, "limit": MAX_INDEXES, "remaining": remaining, "count": len(used_metas[str(p["projectId"])])})
199 return results
202def get_colname_by_key(project_id, key):
203 if key is None or len(key) == 0:
204 return None
206 meta_keys = get(project_id=project_id)
207 meta_keys = {m["key"]: m["index"] for m in meta_keys if m["key"] == key}
208 if len(meta_keys) == 0:
209 return None
211 return index_to_colname(meta_keys[key])
214cache = TTLCache(maxsize=1000, ttl=120)
217@cached(cache)
218def get_for_filters(project_id):
219 with pg_client.PostgresClient() as cur:
220 query = cur.mogrify(f"""SELECT {",".join(column_names())}
221 FROM public.projects
222 WHERE project_id = %(project_id)s
223 AND deleted_at ISNULL
224 LIMIT 1;""", {"project_id": project_id})
225 cur.execute(query=query)
226 metas = cur.fetchone()
227 results = []
228 if metas is not None:
229 for i, k in enumerate(metas.keys()):
230 if metas[k] is not None:
231 results.append({"id": f"meta_{i}",
232 "name": k,
233 "displayName": metas[k],
234 "possibleTypes": ["string"],
235 "dataType": "string",
236 "autoCaptured": True})
237 return {
238 "total": len(results),
239 "displayName": "Metadata",
240 "scope": ["sessions"],
241 "list": results
242 }