query builder

app/query_builder.py

@@ -0,0 +1,122 @@
+# Example usage
+
+# def _example():
+#     READ_PARAMS = "column1, column2"
+#     map_md5 = nullable("some_md5_value")  # Use nullable(None) to include NULL
+#     user_id = 123
+#     mods = 0
+#     status = 2
+#     mode = 0
+#     page_size = None
+#     page = 1
+
+#     query, params = build(
+#         f"SELECT {READ_PARAMS} FROM scores WHERE 1 = 1",
+#         (map_md5, "AND map_md5 = :map_md5"),
+#         (mods, "AND mods = :mods"),
+#         (status, "AND status = :status"),
+#         (mode, "AND mode = :mode"),
+#         (user_id, "AND userid = :userid"),
+#         (
+#             (page_size, "LIMIT :page_size"),
+#             lambda: (
+#                 (page - 1) * page_size if page is not None else None,
+#                 "OFFSET :offset",
+#             ),
+#         ),
+#     )
+
+#     print(query)  # Output the constructed query
+#     print(params)  # Output the parameters dictionary
+
+# _example()
+from typing import Tuple, Dict, Optional, Callable, Union
+
+DatabaseAllowedNotNull = Union[str, int, bool, float]
+Value = Union[DatabaseAllowedNotNull, None]
+SQLValueWithTemplate = Tuple[Value, str]
+SQLValueWithNested = Tuple[Value, "SQLPart"]
+SQLType = Union[SQLValueWithTemplate, SQLValueWithNested]
+SQLPart = Union[SQLType, Tuple[SQLType, ...], Callable[[], SQLType]]
+
+
+class Nullable:
+    def __init__(self, value: Value):
+        self.value = value
+
+
+def nullable(value: DatabaseAllowedNotNull | None) -> Nullable:
+    return Nullable(value)
+
+
+def build(
+    *parts: SQLPart | str,
+) -> Tuple[str, Dict[str, Value]]:
+    parameters = {}
+    query_parts = (_process_query_part(p, parameters) for p in parts)
+
+    query = " ".join(q for q in query_parts if q is not None)
+    return query, parameters
+
+
+def _is_nullable(value) -> bool:
+    return isinstance(value, Nullable)
+
+
+def _extract_value(value) -> Value:
+    if _is_nullable(value):
+        return value.value
+    return value
+
+
+def _process_query_part(
+    part: SQLPart | str, parameters: Dict[str, Value]
+) -> Optional[str]:
+    # late evaluation
+    if callable(part):
+        part = part()
+
+    match part:
+        case None:
+            return None
+        case str(literal):
+            return literal
+        case tuple(parts):
+            return _process_tuple_part(parts, parameters)
+
+    raise TypeError(f"Unexpected type for query part: {type(part)}")
+
+
+def _process_tuple_part(
+    part: SQLPart,
+    parameters: Dict[str, Value],
+) -> Optional[str]:
+    match part:
+        case (None, *_):
+            return None
+        case (
+            Nullable(value=cond) | bool(cond) | str(cond) | int(cond) | float(cond),
+            val,
+        ):
+            evaluated = _process_query_part(val, parameters)
+
+            revealed_cond = _extract_value(cond)
+            if revealed_cond is not None or _is_nullable(cond):
+                if ":" in evaluated and revealed_cond is not None:
+                    parameter_name = (
+                        evaluated.split(":")[-1].strip().split(" ")[0].strip()
+                    )
+                    parameters[parameter_name] = revealed_cond
+            return evaluated
+
+        case tuple(_), *_:  # head, tails
+            parts: list[str] = []
+            for elem in part:
+                return_value = _process_query_part(elem, parameters)
+                if return_value is None:
+                    return None
+                parts.append(return_value)
+            return " ".join(parts)
+
+        case _:
+            raise TypeError(f"Unexpected type for query part: {type(part)} {part}")