zdirpgsql.pas

控件名称：WINZEOS.ZIP 2002年08月03日 作者：CapellaDevelopment Zeos系列构件套件包含一组可直接存取Linux知名的免费数据库MySQL、Postgre

{********************************************************}
{                                                        }
{                 Zeos Database Objects                  }
{              PostgreSql direct class API               }
{                                                        }
{       Copyright (c) 1999-2001 Sergey Seroukhov         }
{    Copyright (c) 1999-2001 Zeos Development Group      }
{                                                        }
{********************************************************}

unit ZDirPgSql;

interface

uses SysUtils, Classes, ZDirSql, ZLibPgSql, DB, ZTransact, ZSqlTypes,
  ZSqlExtra, ZToken;

{$IFNDEF LINUX}
{$INCLUDE ..\Zeos.inc}
{$ELSE}
{$INCLUDE ../Zeos.inc}
{$ENDIF}

type
  { Direct PostgreSql connection }
  TDirPgSqlConnect = class (TDirConnect)
  private
    FError: string;
  protected
    function GetErrorMsg: ShortString; override;
  public
    constructor Create;
    destructor  Destroy; override;

    procedure Connect; override;
    procedure Disconnect; override;
    procedure CreateDatabase(Params: string); override;
    procedure DropDatabase; override;

    function GetConnectStr(Db: string): string;
  end;

  { Transaction types }
  TZPgSqlTransIsolation = (ptDefault, ptReadCommitted, ptRepeatableRead);

  { Direct PostgreSql transaction }
  TDirPgSqlTransact = class (TDirTransact)
  private
    FHandle: PPGconn;
    FError: string;
    FTypeList: TStringList;
    FNotice: AnsiString;
    FTransIsolation: TZPgSqlTransIsolation;
  protected
    function GetErrorMsg: ShortString; override;
    function GetTypeName(TypeNum: Oid): ShortString;
    function GetPid: Integer;
    function GetStatus: TDirStatus; override;
  public
    constructor Create(AConnect: TDirPgSqlConnect);
    destructor Destroy; override;

    procedure Open; override;
    procedure Close; override;
    procedure StartTransaction; override;
    procedure EndTransaction; override;
    procedure Commit; override;
    procedure Rollback; override;

    procedure Reset;

    property Handle: PPGconn read FHandle;
    property Pid: Integer read GetPid;
    property Notice: AnsiString read FNotice write FNotice;
    property TransIsolation: TZPgSqlTransIsolation read FTransIsolation
      write FTransIsolation;
  end;

  { Direct PostgreSql query }
  TDirPgSqlQuery = class (TDirQuery)
  protected
    FHandle: PPGresult;
    FLastInsertOid: Oid;
    FCursorName: string;
    function GetErrorMsg: ShortString; override;
  public
    constructor Create(AConnect: TDirPgSqlConnect; ATransact: TDirPgSqlTransact);

    function  Execute: LongInt; override;
    procedure Open; override;
    procedure Close; override;
    function  CreateBlobObject: TDirBlob; override;

    procedure First; override;
    procedure Last; override;
    procedure Prev; override;
    procedure Next; override;
    procedure Go(Num: Integer); override;

    procedure ShowDatabases(DatabaseName: ShortString); override;
    procedure ShowTables(TableName: ShortString); override;
    procedure ShowColumns(TableName, ColumnName: ShortString); override;
    procedure ShowIndexes(TableName: ShortString); override;

    function  FieldCount: Integer; override;
    function  RecordCount: Integer; override;

    function  FieldName(FieldNum: Integer): ShortString; override;
    function  FieldSize(FieldNum: Integer): Integer; override;
    function  FieldMaxSize(FieldNum: Integer): Integer; override;
    function  FieldType(FieldNum: Integer): Integer; override;
    function  FieldDataType(FieldNum: Integer): TFieldType; override;
    function  FieldIsNull(FieldNum: Integer): Boolean; override;
    function  Field(FieldNum: Integer): string; override;
    function  FieldBuffer(FieldNum: Integer): PChar; override;

    function  FieldMinSize(FieldNum: Integer): Integer;
    function  FieldTypeName(FieldNum: Integer): ShortString;
    function  StringToSql(Value: string): string; override;

    property  Handle: PPGresult read FHandle;
    property  LastInsertOid: Oid read FLastInsertOid;
    property  CursorName: string read FCursorName write FCursorName;
  end;

  { PostgreSql large object }
  TDirPgSqlBlob = class (TDirBlob)
  private
    FBlobHandle: Integer;
  protected
    function GetPosition: LongInt; override;
  public
    constructor Create(AConnect: TDirPgSqlConnect; ATransact: TDirPgSqlTransact;
      AHandle: TBlobHandle);

    procedure Open(Mode: Integer); override;
    procedure Close; override;
    procedure CreateBlob; override;
    procedure DropBlob; override;

    function  Read(Buffer: PChar; Length: Integer): Integer; override;
    function  Write(Buffer: PChar; Length: Integer): Integer; override;
    procedure Seek(Offset: LongInt; Origin: Integer); override;

    procedure ImportFile(FileName: ShortString); override;
    procedure ExportFile(FileName: ShortString); override;

    property BlobHandle: Integer read FBlobHandle;
  end;

  { PostgreSQL array class }
  (*
  TDirPgSqlArray = class(TDirArray)
    protected
      function GetAsString: string; override;
      procedure SetAsString(Value: string); override;
    public
      function Slice(Start, End: Integer) Variant;
      function Value: Variant;
      function
  end;
  *)

  { PostgreSQL class for asynchrounous notifying}
  TDirPgSqlNotify = class(TDirNotify)
  protected
    FHandle: PPGnotify;
    FQuery: TDirPgSqlQuery;
    procedure InternalExec(Sql: string);
  public
    constructor Create(AConnect: TDirPgSqlConnect; ATransact: TDirPgSqlTransact);
    destructor Destroy; override;

    procedure ListenTo(Event: string); override;
    procedure UnlistenTo(Event: string); override;
    procedure DoNotify(Event: string); override;
    function CheckEvents: string; override;

    property Handle: PPGnotify read fHandle;
  end;

{ Convert postgresql field types to delphi field types }
function PgSqlToDelphiType(Value: string; var Size: Integer;
  var ArraySubType: TFieldType; var BlobType: TBlobType): TFieldType;

{ Monitor list }
var
  MonitorList: TZMonitorList;

implementation

uses ZDBaseConst, ZExtra {$IFNDEF LINUX}, ActiveX{$ENDIF};

{***************** TDirPgSqlConnect implementation *****************}

{ Class constructor }
constructor TDirPgSqlConnect.Create;
begin
  inherited Create;
  Port := '5432';
end;

{ Class destructor }
destructor TDirPgSqlConnect.Destroy;
begin
  inherited Destroy;
end;

{ Get an error message }
function TDirPgSqlConnect.GetErrorMsg: ShortString;
begin
  if Status <> csOk then
    Result := FError
  else
    Result := '';
end;

{ Construct postgresql connect string }
function TDirPgSqlConnect.GetConnectStr(Db: string): string;

  function CheckAddr(Value: string): Boolean;
  var
    I, N: Integer;
  begin
    Result := False;
    N := 0;
    for I := 1 to Length(Value) do
    begin
      if Value[I] = '.' then
        Inc(N)
      else if not (Value[I] in ['0'..'9']) then
        Exit;
    end;
    Result := (N = 3);
  end;

begin
  HostName := Trim(HostName);
  if CheckAddr(HostName) then
    Result := 'hostaddr='
  else
    Result := 'host=';
  Result := Result + HostName + ' port=' + Port +
    ' dbname=' + Db + ' user=' + Login + ' password=' + Passwd;
end;

{ Connect to database }
procedure TDirPgSqlConnect.Connect;
begin
  inherited Connect;
  if hDll = 0 then PgSqlLoadLib;
  SetStatus(csOk);
  SetActive(True);
end;

{ Disconnect from database }
procedure TDirPgSqlConnect.Disconnect;
begin
  SetStatus(csOk);
  SetActive(False);
end;

{ Create and connect to database }
procedure TDirPgSqlConnect.CreateDatabase(Params: string);
var
  Handle: PPGconn;
  Result: PPGresult;
  Buffer: string;
begin
  if Active then Disconnect;
  if hDll = 0 then PgSqlLoadLib;
  SetStatus(csFail);
  FError := SDbCreateError;

  Handle := PQconnectdb(PChar(GetConnectStr('template1')));

  if not Assigned(Handle) then Exit;

  Buffer := 'CREATE DATABASE ' + Database + ' ' + Params;
  Result := PQexec(Handle, PChar(Buffer));
  if Assigned(Result) then
  begin
    SetStatus(csOk);
    PQclear(Result);
  end;
  MonitorList.InvokeEvent(Buffer, 'Fail.', Status <> csOk);
end;

{ Drop current database }
procedure TDirPgSqlConnect.DropDatabase;
var
  Handle: PPGconn;
  Result: PPGresult;
  Buffer: string;
begin
  if Active then Disconnect;
  SetStatus(csFail);
  FError := SConnectError;

  Buffer := 'DROP DATABASE ' + Database;
  Handle := PQconnectdb(PChar(GetConnectStr('template1')));
  if Assigned(Handle) then
  begin
    Result := PQexec(Handle, PChar(Buffer));
    if Assigned(Result) then
      SetStatus(csOk);
    PQclear(Result);
    PQfinish(Handle);
  end;

  MonitorList.InvokeEvent(Buffer, 'Fail.', Status <> csOk);
end;

{*********** TDirPgSqlTransact implementation *********}

{ NoticeProcessor handler }
procedure NoticeProc(Arg: Pointer; Msg: PChar); cdecl;
begin
  if Assigned(Arg) then
    TDirPgSqlTransact(Arg).Notice := TDirPgSqlTransact(Arg).Notice + Msg;
end;

{ Class constructor }
constructor TDirPgSqlTransact.Create(AConnect: TDirPgSqlConnect);
begin
  inherited Create;
  Connect := AConnect;
  FHandle := nil;
end;

{ Class destructor }
destructor TDirPgSqlTransact.Destroy;
begin
  inherited Destroy;
  if Assigned(FTypeList) then
    FTypeList.Free;
end;

{ Get error message }
function TDirPgSqlTransact.GetErrorMsg: ShortString;
begin
  if Assigned(Handle) and Assigned(Connect) then
    FError := Trim(StrPas(PQerrorMessage(Handle)));
  Result := FError;
end;

{ Get field type by oid }
function TDirPgSqlTransact.GetTypeName(TypeNum: Oid): ShortString;
var
  I: Integer;
  AResult: PPGresult;
begin
  Result := '';
  if not Assigned(Handle) then Exit;
  if not Assigned(FTypeList) then
  begin
    FTypeList := TStringList.Create;
    AResult := PQexec(Handle, 'SELECT oid, typname FROM pg_type WHERE oid<10000');
    if not Assigned(AResult) then Exit;
    for I := 0 to PQntuples(AResult)-1 do
      FTypeList.AddObject(StrPas(PQgetvalue(AResult,I,1)),
        TObject(StrToIntDef(StrPas(PQgetvalue(AResult,I,0)),0)));
    PQclear(AResult);
  end;
  I := FTypeList.IndexOfObject(TObject(TypeNum));
  if I >= 0 then
    Result := FTypeList.Strings[I];
end;

{ Retrieve backend server's process id (PID }
function TDirPgSqlTransact.GetPID: Integer;
begin
  Result := PQbackendPID(Handle);
end;

{ Get status directly from the PQlib interface}
function TDirPgSqlTransact.GetStatus: TDirStatus;
begin
  Result := inherited GetStatus;
  if Handle <> nil then
  begin
     case PQstatus(Handle) of
       CONNECTION_OK:
         Result := csOK;
       CONNECTION_BAD:
         Result := csFail
     end;
  end;
end;

{ Connect to database }
procedure TDirPgSqlTransact.Open;
begin
  inherited Open;
  SetStatus(csFail);
  if not Assigned(Connect) or not Connect.Active then
    Exit;

  FHandle := PQconnectdb(PChar(TDirPgSqlConnect(Connect).
    GetConnectStr(Connect.Database)));

  if PQstatus(Handle) = CONNECTION_BAD then
  begin
    if not Assigned(Handle) then
      FError := SConnectError;
    MonitorList.InvokeEvent(Format('CONNECT %s',[Connect.Database]), FError, True);
    Exit;
  end;
  MonitorList.InvokeEvent(Format('CONNECT %s',[Connect.Database]), 'OK.', False);
  StartTransaction;
  PQsetNoticeProcessor(FHandle, NoticeProc, Self);
  SetActive(Status = csOk);
end;

{ Disconnect from database }
procedure TDirPgSqlTransact.Close;
begin
  EndTransaction;
  if Active then
  begin
    PQfinish(Handle);
    FHandle := nil;
    if Assigned(FTypeList) then
    begin
      FTypeList.Free;
      FTypeList := nil;
    end;
    if Assigned(Connect) then
      MonitorList.InvokeEvent(Format('DISCONNECT %s',[Connect.Database]), 'OK.', False);
  end;
  SetActive(False);
end;

{ Start transaction }
procedure TDirPgSqlTransact.StartTransaction;
var
  Result: PPGresult;