nur-sery/nur_src/m/pgsql/PgsqlConn.php

<?php
namespace nur\m\pgsql;

use nur\config;
use nur\debug;
use nur\m\base\AbstractConn;
use nur\m\base\QueryException;
use nur\m\IQuery;
use nur\m\IRowIncarnation;
use nur\md;

class PgsqlConn extends AbstractConn {
  static function query_exception($res=null, ?string $cause=null): QueryException {
    $parts = [];
    if ($cause) $parts[] = $cause;
    if ($res !== null) $parts[] = pg_last_error($res);
    elseif (!$parts) $parts[] = "unknown error";
    $message = implode(": ", $parts);
    return new QueryException($message);
  }

  protected $dbname, $dbuser, $dbpass, $options;
  protected $conn;

  protected $inTransaction = false;

  function __construct($dbname, ?string $dbuser=null, ?string $dbpass=null) {
    if (is_array($dbname)) {
      if ($dbuser === null) $dbuser = $dbname["user"];
      if ($dbpass === null) $dbpass = $dbname["pass"];
      $dbname = $dbname["name"];
    }
    if (strpos($dbname, "=") === false) {
      $connString = "dbname=$dbname";
    } else {
      $connString = $dbname;
    }
    if ($dbuser !== null) $connString .= " user=$dbuser";
    if ($dbpass !== null) $connString .= " password=$dbpass";
    $this->dbname = $dbname;
    $this->dbuser = $dbuser;
    $this->dbpass = $dbpass;
    $conn = pg_connect($connString);
    if ($conn === false) {
      throw self::query_exception(null, "unable to connect");
    }
    $this->conn = $conn;
  }

  function __destruct() {
    $this->rollback();
    if ($this->conn) pg_close($this->conn);
  }

  function getInfos(): array {
    return [$this->dbname, $this->dbuser, $this->dbpass];
  }

  function beginTransaction(): void {
    if (!$this->inTransaction) {
      if (pg_query($this->conn, "begin") === false) {
        throw self::query_exception(null, "unable to begin transaction");
      }
      $this->inTransaction = true;
    }
  }

  function commit(): void {
    if ($this->inTransaction) {
      $this->inTransaction = false;
      if (pg_query($this->conn, "commit") === false) {
        throw self::query_exception($this->conn, "error on commit");
      }
    }
  }

  function rollback(): void {
    if ($this->inTransaction) {
      $this->inTransaction = false;
      if (pg_query($this->conn, "rollback") === false) {
        throw self::query_exception($this->conn, "error on rollback");
      }
    }
  }

  function _execute0(string $sql, ?array &$bindings, array $params): array {
    if ($params["transaction"]) $this->beginTransaction();

    if (!$bindings) {
      $stmt = pg_query($this->conn, $sql);
    } else {
      $bparams = [];
      $number = 1;
      foreach ($bindings as $name => $value) {
        $sql = str_replace(":$name", "\$$number", $sql);
        $bparams[] = $value;
        $number++;
      }
      $stmt = pg_query_params($this->conn, $sql, $bparams);
    }
    if ($stmt === false) {
      throw self::query_exception($this->conn, "execute error");
    }

    $r = [];
    if ($params["num_rows"]) {
      $numRows = pg_affected_rows($stmt);
      if ($numRows === false) throw self::query_exception($stmt, "error getting num_rows");
      $r["num_rows"] = $numRows;
    }
    if ($params["last_insert_id"]) {
      $r["insert_id"] = null;
    }
    if ($params["stmt"]) {
      $r["stmt"] = $stmt;
    } else {
      pg_free_result($stmt);
    }
    return $r;
  }

  function _prepareLogger(string $sql, ?array $bindings): array {
    $queryLogger = $this->queryLogger;
    $actualQuery = null;
    $traceSql = config::k("trace_sql", false);
    if ($traceSql || $queryLogger !== null) {
      $actualQuery = PgsqlQuery::build_actual_query($sql, $bindings);
    }
    if ($traceSql) debug::log("SQL TRACE --", $actualQuery);
    return [$queryLogger, $actualQuery];
  }

  function _execute1(string $sql, ?array &$bindings, array $params): array {
    [$queryLogger, $actualQuery] = $this->_prepareLogger($sql, $bindings);
    $r = $this->_execute0($sql, $bindings, $params);
    if ($queryLogger !== null) $queryLogger->logQuery($actualQuery);
    return $r;
  }

  function _execute(string $sql, ?array &$bindings=null, ?array $params=null): array {
    md::ensure_schema($params, self::EXECUTE_PARAMS_SCHEMA);
    return $this->_execute1($sql, $bindings, $params);
  }

  function _fetchAll(string $sql, ?array &$bindings=null): array {
    ["stmt" => $stmt
    ] = $this->_execute1($sql, $bindings, self::EXECUTE_PARAMS_DQL);
    try {
      $rows = pg_fetch_all($stmt, PGSQL_ASSOC);
      return $rows;
    } finally {
      pg_free_result($stmt);
    }
  }

  function _fetchFirst(string $sql, ?array &$bindings=null): ?array {
    ["stmt" => $stmt
    ] = $this->_execute1($sql, $bindings, self::EXECUTE_PARAMS_DQL);
    $row = pg_fetch_assoc($stmt);
    pg_free_result($stmt);
    return $row !== false? $row: null;
  }

  function _update(string $sql, ?array &$bindings=null): int {
    ["num_rows" => $numRows
    ] = $this->_execute1($sql, $bindings, self::EXECUTE_PARAMS_DML_UPDATE);
    return $numRows;
  }

  function query(?string $sql=null, ?array $filter=null, ?IRowIncarnation $incarnation=null): IQuery {
    return new PgsqlQuery($this, $sql, $filter, $incarnation);
  }
}