sql-query-builder

SQL Query Builder

Overview

Based on DDC methodology (Chapter 2.1), this skill builds SQL queries for construction databases, supporting common construction data patterns like cost tracking, schedule management, and resource allocation.

Book Reference: "Типы данных в строительстве" / "Data Types in Construction"

Quick Start

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Any, Union
from datetime import datetime, date

class QueryType(Enum):
    """Types of SQL queries"""
    SELECT = "select"
    INSERT = "insert"
    UPDATE = "update"
    DELETE = "delete"
    AGGREGATE = "aggregate"
    JOIN = "join"

class AggregateFunction(Enum):
    """SQL aggregate functions"""
    SUM = "SUM"
    AVG = "AVG"
    COUNT = "COUNT"
    MIN = "MIN"
    MAX = "MAX"
    GROUP_CONCAT = "GROUP_CONCAT"

class JoinType(Enum):
    """SQL join types"""
    INNER = "INNER JOIN"
    LEFT = "LEFT JOIN"
    RIGHT = "RIGHT JOIN"
    FULL = "FULL OUTER JOIN"

class ComparisonOperator(Enum):
    """Comparison operators"""
    EQ = "="
    NE = "<>"
    GT = ">"
    GE = ">="
    LT = "<"
    LE = "<="
    LIKE = "LIKE"
    IN = "IN"
    BETWEEN = "BETWEEN"
    IS_NULL = "IS NULL"
    IS_NOT_NULL = "IS NOT NULL"

@dataclass
class WhereCondition:
    """SQL WHERE condition"""
    column: str
    operator: ComparisonOperator
    value: Any
    table_alias: Optional[str] = None

    def to_sql(self) -> str:
        col = f"{self.table_alias}.{self.column}" if self.table_alias else self.column

        if self.operator == ComparisonOperator.IS_NULL:
            return f"{col} IS NULL"
        elif self.operator == ComparisonOperator.IS_NOT_NULL:
            return f"{col} IS NOT NULL"
        elif self.operator == ComparisonOperator.IN:
            values = ", ".join(f"'{v}'" if isinstance(v, str) else str(v) for v in self.value)
            return f"{col} IN ({values})"
        elif self.operator == ComparisonOperator.BETWEEN:
            return f"{col} BETWEEN {self._format_value(self.value[0])} AND {self._format_value(self.value[1])}"
        elif self.operator == ComparisonOperator.LIKE:
            return f"{col} LIKE '{self.value}'"
        else:
            return f"{col} {self.operator.value} {self._format_value(self.value)}"

    def _format_value(self, value: Any) -> str:
        if isinstance(value, str):
            return f"'{value}'"
        elif isinstance(value, (date, datetime)):
            return f"'{value.isoformat()}'"
        elif value is None:
            return "NULL"
        else:
            return str(value)

@dataclass
class JoinClause:
    """SQL JOIN clause"""
    table: str
    alias: str
    join_type: JoinType
    on_left: str
    on_right: str

    def to_sql(self) -> str:
        return f"{self.join_type.value} {self.table} {self.alias} ON {self.on_left} = {self.on_right}"

@dataclass
class SelectColumn:
    """Column selection with optional aggregate and alias"""
    column: str
    table_alias: Optional[str] = None
    aggregate: Optional[AggregateFunction] = None
    alias: Optional[str] = None

    def to_sql(self) -> str:
        col = f"{self.table_alias}.{self.column}" if self.table_alias else self.column

        if self.aggregate:
            col = f"{self.aggregate.value}({col})"

        if self.alias:
            col = f"{col} AS {self.alias}"

        return col

@dataclass
class OrderBy:
    """ORDER BY clause"""
    column: str
    descending: bool = False
    table_alias: Optional[str] = None

    def to_sql(self) -> str:
        col = f"{self.table_alias}.{self.column}" if self.table_alias else self.column
        direction = "DESC" if self.descending else "ASC"
        return f"{col} {direction}"


class ConstructionQueryBuilder:
    """
    Build SQL queries for construction databases.
    Based on DDC methodology Chapter 2.1.
    """

    def __init__(self, dialect: str = "postgresql"):
        self.dialect = dialect
        self.schemas = self._define_construction_schemas()

    def _define_construction_schemas(self) -> Dict[str, Dict]:
        """Define common construction database schemas"""
        return {
            "projects": {
                "columns": ["id", "name", "code", "status", "start_date", "end_date", "budget", "client_id"],
                "primary_key": "id"
            },
            "cost_items": {
                "columns": ["id", "project_id", "wbs_code", "description", "budgeted_cost", "actual_cost", "committed_cost"],
                "primary_key": "id",
                "foreign_keys": {"project_id": "projects.id"}
            },
            "activities": {
                "columns": ["id", "project_id", "name", "wbs_code", "start_date", "end_date", "duration", "status", "percent_complete"],
                "primary_key": "id",
                "foreign_keys": {"project_id": "projects.id"}
            },
            "resources": {
                "columns": ["id", "name", "type", "rate", "unit"],
                "primary_key": "id"
            },
            "resource_assignments": {
                "columns": ["id", "activity_id", "resource_id", "units", "cost"],
                "primary_key": "id",
                "foreign_keys": {"activity_id": "activities.id", "resource_id": "resources.id"}
            },
            "change_orders": {
                "columns": ["id", "project_id", "number", "description", "amount", "status", "submitted_date", "approved_date"],
                "primary_key": "id",
                "foreign_keys": {"project_id": "projects.id"}
            },
            "invoices": {
                "columns": ["id", "project_id", "number", "amount", "status", "invoice_date", "due_date", "paid_date"],
                "primary_key": "id",
                "foreign_keys": {"project_id": "projects.id"}
            },
            "materials": {
                "columns": ["id", "name", "category", "unit", "unit_cost"],
                "primary_key": "id"
            },
            "material_requisitions": {
                "columns": ["id", "project_id", "material_id", "quantity", "required_date", "status"],
                "primary_key": "id",
                "foreign_keys": {"project_id": "projects.id", "material_id": "materials.id"}
            },
            "daily_reports": {
                "columns": ["id", "project_id", "report_date", "weather", "temperature", "crew_count", "notes"],
                "primary_key": "id",
                "foreign_keys": {"project_id": "projects.id"}
            }
        }

    def select(
        self,
        table: str,
        columns: List[Union[str, SelectColumn]],
        conditions: Optional[List[WhereCondition]] = None,
        order_by: Optional[List[OrderBy]] = None,
        limit: Optional[int] = None,
        offset: Optional[int] = None
    ) -> str:
        """Build a SELECT query"""
        # Format columns
        cols = []
        for col in columns:
            if isinstance(col, str):
                cols.append(col)
            else:
                cols.append(col.to_sql())

        query = f"SELECT {', '.join(cols)}\nFROM {table}"

        # WHERE clause
        if conditions:
            where_parts = [c.to_sql() for c in conditions]
            query += f"\nWHERE {' AND '.join(where_parts)}"

        # ORDER BY
        if order_by:
            order_parts = [o.to_sql() for o in order_by]
            query += f"\nORDER BY {', '.join(order_parts)}"

        # LIMIT/OFFSET
        if limit:
            query += f"\nLIMIT {limit}"
        if offset:
            query += f"\nOFFSET {offset}"

        return query + ";"

    def select_with_joins(
        self,
        main_table: str,
        main_alias: str,
        columns: List[SelectColumn],
        joins: List[JoinClause],
        conditions: Optional[List[WhereCondition]] = None,
        group_by: Optional[List[str]] = None,
        having: Optional[List[WhereCondition]] = None,
        orde