ds_provider_postgresql_py_lib.linked_service¶

File: __init__.py Region: ds_provider_postgresql_py_lib/linked_service

PostgreSQL Linked Service

This module implements a linked service for PostgreSQL databases.

Example

>>> linked_service = PostgreSQLLinkedService(
...     settings=PostgreSQLLinkedServiceSettings(
...         uri="postgresql://user:password@localhost:5432/mydb",
...     ),
... )
>>> linked_service.connect()

Submodules¶

ds_provider_postgresql_py_lib.linked_service.postgresql

Classes¶

`PostgreSQLLinkedService`	The class is used to connect with PostgreSQL database.
`PostgreSQLLinkedServiceSettings`	The object containing the PostgreSQL linked service settings.

Package Contents¶

class ds_provider_postgresql_py_lib.linked_service.PostgreSQLLinkedService[source]¶

Bases: ds_resource_plugin_py_lib.common.resource.linked_service.LinkedService[PostgreSQLLinkedServiceSettingsType], Generic[PostgreSQLLinkedServiceSettingsType]

The class is used to connect with PostgreSQL database.

settings: PostgreSQLLinkedServiceSettingsType¶

_engine: sqlalchemy.Engine | None = None¶: The SQLAlchemy engine instance with connection pool.

property type: ds_provider_postgresql_py_lib.enums.ResourceType¶

Get the type of the linked service.

Returns:: ResourceType

property engine: sqlalchemy.Engine | None¶

Get the SQLAlchemy engine instance.

Returns:: The engine if initialized, None otherwise.
Return type:: Engine | None

property connection: sqlalchemy.Engine¶

Get the established backend connection object.

Returns:: The initialized SQLAlchemy engine.
Return type:: Engine
Raises:: ConnectionError – If connect() has not been called successfully.

property pool: sqlalchemy.pool.Pool | None¶

Get the connection pool from the engine.

The pool is automatically created by SQLAlchemy when create_engine() is called with pool parameters. All connections (via engine.connect(), engine.begin(), etc.) automatically use this pool.

Returns:: The connection pool if the engine is initialized, None otherwise.
Return type:: Pool | None

connect() → None[source]¶

Connect to the PostgreSQL database and create a connection pool.

Returns:: None

test_connection() → tuple[bool, str][source]¶

Test the connection to the PostgreSQL database.

Returns:: A tuple containing a boolean indicating success and a string message.
Return type:: tuple[bool, str]

_is_authentication_error(error: Exception) → bool[source]¶

Detect authentication failures from SQLAlchemy/driver exceptions.

Parameters:: error – The exception to check.
Returns:: True if the error is an authentication error, False otherwise.
Return type:: bool

close() → None[source]¶: Close the linked service.

class ds_provider_postgresql_py_lib.linked_service.PostgreSQLLinkedServiceSettings[source]¶

Bases: ds_resource_plugin_py_lib.common.resource.linked_service.LinkedServiceSettings

The object containing the PostgreSQL linked service settings.

uri: str¶

PostgreSQL connection URI.

Format: postgresql://[user[:password]@][host][:port][/database][?param1=value1&param2=value2]

Examples

postgresql://user:password@localhost:5432/mydb postgresql://user:password@localhost:5432/mydb?sslmode=require postgresql://user@localhost/mydb postgresql://localhost/mydb

pool_size: int = 5¶: The size of the connection pool. Defaults to 5.

max_overflow: int = 10¶: The maximum overflow connections allowed. Defaults to 10.

pool_timeout: int = 30¶: The timeout in seconds for getting a connection from the pool. Defaults to 30.

pool_recycle: int = 3600¶: The time in seconds after which a connection is recycled. Defaults to 3600.