Source code for ace.database.query_builder

from abc import ABC, abstractmethod
from typing import Union


[docs] class Condition:
[docs] def __init__(self, column: str, value: Union[str, int, None], operator: str = "="): self.column = column self.value = value self.operator = operator
def __str__(self): if self.value is None: return f"{self.column} IS NULL" return f"{self.column} {self.operator} '{self.value}'"
[docs] class BaseQueryBuilder(ABC):
[docs] @abstractmethod def build_query(self, *args, **kwargs): pass
[docs] def __init__(self, table_name: str, columns: Union[list, str] = "*"): self.table_name = table_name self.columns = columns
[docs] class SelectQueryBuilder(BaseQueryBuilder):
[docs] def build_query(self, order_by: str = None, limit: int = None, offset: int = None, conditions: list[Condition] = None): query = f"SELECT DISTINCT {','.join(self.columns)} FROM {self.table_name}" if conditions: query += " WHERE " query += " AND ".join( [condition.__str__() for condition in conditions] ) if order_by: query += f" ORDER BY {order_by}" if limit: query += f" LIMIT {limit}" if offset: query += f" OFFSET {offset}" return query