Class AlloyDBEngine (0.5.0)

AlloyDBEngine(
    key: object,
    engine: sqlalchemy.ext.asyncio.engine.AsyncEngine,
    loop: typing.Optional[asyncio.events.AbstractEventLoop],
    thread: typing.Optional[threading.Thread],
)

A class for managing connections to a AlloyDB database.

Methods

AlloyDBEngine

AlloyDBEngine(
    key: object,
    engine: sqlalchemy.ext.asyncio.engine.AsyncEngine,
    loop: typing.Optional[asyncio.events.AbstractEventLoop],
    thread: typing.Optional[threading.Thread],
)

AlloyDBEngine constructor.

Parameters
Name Description
key object

Prevent direct constructor usage.

engine AsyncEngine

Async engine connection pool.

loop Optional[asyncio.AbstractEventLoop]

Async event loop used to create the engine.

thread Optional[Thread] = None

Thread used to create the engine async.

Exceptions
Type Description
Exception If the constructor is called directly by the user.

_aexecute

_aexecute(query: str, params: typing.Optional[dict] = None) -> None

Execute a SQL query.

_aexecute_outside_tx

_aexecute_outside_tx(query: str) -> None

Execute a SQL query.

_afetch

_afetch(
    query: str, params: typing.Optional[dict] = None
) -> typing.Sequence[sqlalchemy.engine.row.RowMapping]

Fetch results from a SQL query.

_aload_table_schema

_aload_table_schema(table_name: str) -> sqlalchemy.sql.schema.Table

Load table schema from existing table in PgSQL database.

Returns
Type Description
(sqlalchemy.Table) The loaded table.

_create

_create(
    project_id: str,
    region: str,
    cluster: str,
    instance: str,
    database: str,
    ip_type: typing.Union[str, google.cloud.alloydb.connector.enums.IPTypes],
    user: typing.Optional[str] = None,
    password: typing.Optional[str] = None,
    loop: typing.Optional[asyncio.events.AbstractEventLoop] = None,
    thread: typing.Optional[threading.Thread] = None,
    iam_account_email: typing.Optional[str] = None,
) -> langchain_google_alloydb_pg.engine.AlloyDBEngine

Create an AlloyDBEngine from an AlloyDB instance.

Parameters
Name Description
project_id str

GCP project ID.

region str

Cloud AlloyDB instance region.

cluster str

Cloud AlloyDB cluster name.

instance str

Cloud AlloyDB instance name.

database str

Database name.

ip_type Union[str, IPTypes], optional

IP address type. Defaults to IPTypes.PUBLIC.

user Optional[str], optional

Cloud AlloyDB user name. Defaults to None.

password Optional[str], optional

Cloud AlloyDB user password. Defaults to None.

loop Optional[asyncio.AbstractEventLoop]

Async event loop used to create the engine.

thread Optional[Thread] = None

Thread used to create the engine async.

iam_account_email Optional[str], optional

IAM service account email.

Exceptions
Type Description
ValueError Raises error if only one of 'user' or 'password' is specified.
Returns
Type Description
AlloyDBEngine A newly created AlloyDBEngine instance.

_execute

_execute(query: str, params: typing.Optional[dict] = None) -> None

Execute a SQL query.

_fetch

_fetch(
    query: str, params: typing.Optional[dict] = None
) -> typing.Sequence[sqlalchemy.engine.row.RowMapping]

Fetch results from a SQL query.

_run_as_sync

_run_as_sync(
    coro: typing.Awaitable[langchain_google_alloydb_pg.engine.T],
) -> langchain_google_alloydb_pg.engine.T

Run an async coroutine synchronously

afrom_instance

afrom_instance(
    project_id: str,
    region: str,
    cluster: str,
    instance: str,
    database: str,
    user: typing.Optional[str] = None,
    password: typing.Optional[str] = None,
    ip_type: typing.Union[
        str, google.cloud.alloydb.connector.enums.IPTypes
    ] = IPTypes.PUBLIC,
    iam_account_email: typing.Optional[str] = None,
) -> langchain_google_alloydb_pg.engine.AlloyDBEngine

Create an AlloyDBEngine from an AlloyDB instance.

Parameters
Name Description
project_id str

GCP project ID.

region str

Cloud AlloyDB instance region.

cluster str

Cloud AlloyDB cluster name.

instance str

Cloud AlloyDB instance name.

database str

Cloud AlloyDB database name.

user Optional[str], optional

Cloud AlloyDB user name. Defaults to None.

password Optional[str], optional

Cloud AlloyDB user password. Defaults to None.

ip_type Union[str, IPTypes], optional

IP address type. Defaults to IPTypes.PUBLIC.

iam_account_email Optional[str], optional

IAM service account email. Defaults to None.

Returns
Type Description
AlloyDBEngine A newly created AlloyDBEngine instance.

ainit_chat_history_table

ainit_chat_history_table(table_name: str) -> None

Create an AlloyDB table to save chat history messages.

Parameter
Name Description
table_name str

The table name to store chat history.

ainit_document_table

ainit_document_table(
    table_name: str,
    content_column: str = "page_content",
    metadata_columns: typing.List[langchain_google_alloydb_pg.engine.Column] = [],
    metadata_json_column: str = "langchain_metadata",
    store_metadata: bool = True,
) -> None

Create a table for saving of langchain documents. If table already exists, a DuplicateTableError error is thrown.

Parameters
Name Description
table_name str

The PgSQL database table name.

content_column str

Name of the column to store document content. Default: "page_content".

metadata_columns List[Column]

A list of Columns to create for custom metadata. Optional.

metadata_json_column str

The column to store extra metadata in JSON format. Default: "langchain_metadata". Optional.

store_metadata bool

Whether to store extra metadata in a metadata column if not described in 'metadata' field list (Default: True).

ainit_vectorstore_table

ainit_vectorstore_table(
    table_name: str,
    vector_size: int,
    content_column: str = "content",
    embedding_column: str = "embedding",
    metadata_columns: typing.List[langchain_google_alloydb_pg.engine.Column] = [],
    metadata_json_column: str = "langchain_metadata",
    id_column: str = "langchain_id",
    overwrite_existing: bool = False,
    store_metadata: bool = True,
) -> None

Create a table for saving of vectors to be used with AlloyDB. If table already exists and overwrite flag is not set, a TABLE_ALREADY_EXISTS error is thrown.

Parameters
Name Description
table_name str

The table name.

vector_size int

Vector size for the embedding model to be used.

content_column str

Name of the column to store document content. Default: "page_content".

embedding_column str

Name of the column to store vector embeddings. Default: "embedding".

metadata_columns List[Column]

A list of Columns to create for custom metadata. Default: []. Optional.

metadata_json_column str

The column to store extra metadata in JSON format. Default: "langchain_metadata". Optional.

id_column str

Name of the column to store ids. Default: "langchain_id". Optional,

overwrite_existing bool

Whether to drop the existing table before insertion. Default: False.

store_metadata bool

Whether to store metadata in a JSON column if not specified by metadata_columns. Default: True.

Exceptions
Type Description
DuplicateTableErro

from_engine

from_engine(
    engine: sqlalchemy.ext.asyncio.engine.AsyncEngine,
) -> langchain_google_alloydb_pg.engine.AlloyDBEngine

Create an AlloyDBEngine instance from an AsyncEngine.

from_instance

from_instance(
    project_id: str,
    region: str,
    cluster: str,
    instance: str,
    database: str,
    user: typing.Optional[str] = None,
    password: typing.Optional[str] = None,
    ip_type: typing.Union[
        str, google.cloud.alloydb.connector.enums.IPTypes
    ] = IPTypes.PUBLIC,
    iam_account_email: typing.Optional[str] = None,
) -> langchain_google_alloydb_pg.engine.AlloyDBEngine

Create an AlloyDBEngine from an AlloyDB instance.

Parameters
Name Description
project_id str

GCP project ID.

region str

Cloud AlloyDB instance region.

cluster str

Cloud AlloyDB cluster name.

instance str

Cloud AlloyDB instance name.

database str

Database name.

user Optional[str], optional

Cloud AlloyDB user name. Defaults to None.

password Optional[str], optional

Cloud AlloyDB user password. Defaults to None.

ip_type Union[str, IPTypes], optional

IP address type. Defaults to IPTypes.PUBLIC.

iam_account_email Optional[str], optional

IAM service account email. Defaults to None.

Returns
Type Description
AlloyDBEngine A newly created AlloyDBEngine instance.

init_chat_history_table

init_chat_history_table(table_name: str) -> None

Create an AlloyDB table to save chat history messages.

Parameter
Name Description
table_name str

The table name to store chat history.

init_document_table

init_document_table(
    table_name: str,
    content_column: str = "page_content",
    metadata_columns: typing.List[langchain_google_alloydb_pg.engine.Column] = [],
    metadata_json_column: str = "langchain_metadata",
    store_metadata: bool = True,
) -> None

Create a table for saving of langchain documents. If table already exists, a DuplicateTableError error is thrown.

Parameters
Name Description
table_name str

The PgSQL database table name.

content_column str

Name of the column to store document content. Default: "page_content".

metadata_columns List[Column]

A list of Columns to create for custom metadata. Optional.

metadata_json_column str

The column to store extra metadata in JSON format. Default: "langchain_metadata". Optional.

store_metadata bool

Whether to store extra metadata in a metadata column if not described in 'metadata' field list (Default: True).

init_vectorstore_table

init_vectorstore_table(
    table_name: str,
    vector_size: int,
    content_column: str = "content",
    embedding_column: str = "embedding",
    metadata_columns: typing.List[langchain_google_alloydb_pg.engine.Column] = [],
    metadata_json_column: str = "langchain_metadata",
    id_column: str = "langchain_id",
    overwrite_existing: bool = False,
    store_metadata: bool = True,
) -> None

Create a table for saving of vectors to be used with AlloyDB. If table already exists and overwrite flag is not set, a TABLE_ALREADY_EXISTS error is thrown.

Parameters
Name Description
table_name str

The table name.

vector_size int

Vector size for the embedding model to be used.

content_column str

Name of the column to store document content. Default: "page_content".

embedding_column str

Name of the column to store vector embeddings. Default: "embedding".

metadata_columns List[Column]

A list of Columns to create for custom metadata. Default: []. Optional.

metadata_json_column str

The column to store extra metadata in JSON format. Default: "langchain_metadata". Optional.

id_column str

Name of the column to store ids. Default: "langchain_id". Optional,

overwrite_existing bool

Whether to drop the existing table before insertion. Default: False.

store_metadata bool

Whether to store metadata in a JSON column if not specified by metadata_columns. Default: True.

Exceptions
Type Description
DuplicateTableErro