snowshu.adapters.target_adapters.postgres_adapter

class snowshu.adapters.target_adapters.postgres_adapter.postgres_adapter.PostgresAdapter(replica_metadata: dict, **kwargs)

Bases: snowshu.adapters.target_adapters.base_target_adapter.BaseTargetAdapter

BASE_DB_IMAGE = 'postgres:12'

DATA_TYPE_MAPPINGS = {'bigint': bigint, 'binary': binary, 'bit': binary, 'boolean': boolean, 'bytea': binary, 'char': char, 'character': char, 'character_varying': varchar, 'date': date, 'datetime': datetime, 'decimal': decimal, 'double': float, 'double_precision': float, 'float': float, 'float4': float, 'float8': float, 'int': bigint, 'integer': bigint, 'json': json, 'jsonb': json, 'numeric': numeric, 'real': float, 'smallint': bigint, 'string': varchar, 'text': varchar, 'time': time, 'time_with_time_zone': time_tz, 'time_without_time_zone': time, 'timestamp': timestamp_ntz, 'timestamp_ltz': timestamp_tz, 'timestamp_ntz': timestamp_ntz, 'timestamp_tz': timestamp_tz, 'timestamp_with_time_zone': timestamp_tz, 'timestamp_without_time_zone': timestamp_ntz, 'varbinary': binary, 'varchar': varchar}

DEFAULT_CASE = 'lower'

DOCKER_IMAGE = 'postgres:12'

DOCKER_REMOUNT_DIRECTORY = 'snowshu_replica_data'

DOCKER_REPLICA_MOUNT_FOLDER = '/app/snowshu_replica_data_shared'

DOCKER_SNOWSHU_ENVARS = ['POSTGRES_PASSWORD', 'POSTGRES_USER', 'POSTGRES_DB']

MATERIALIZATION_MAPPINGS = {'BASE_TABLE': TABLE, 'TABLE': TABLE, 'VIEW': VIEW}

PRELOADED_PACKAGES = ['postgresql-plpython3-12']

copy_replica_data() → Tuple[bool, str]

A service function that copies replica data to a shared location

create_all_database_extensions() → str

Post-processing step to create extensions on all existing databases

create_database_if_not_exists(database: str) → str

Postgres doesn’t have great CINE support.

So ask for forgiveness instead.

create_or_replace_view(relation) → None

Creates a view of the specified relation in the target adapter.

Relation must have a valid view_ddl value that can be executed as a SELECT statement.

Parameters

relation – the Relation object to be created as a view.

create_schema_if_not_exists(database: str, schema: str) → None

dialect = 'postgresql'

enable_cross_database() → None

Create x-database links, if available to the target.

Parameters

relations – an iterable of relations to collect databases and schemas from.

image_initialize_bash_commands() → List[str]

initialize_replica(source_adapter_name: str, incremental_image: str = None) → None

shimming but will want to move _init_image public with this interface.

Parameters

• source_adapter_name – the classname of the source adapter

• incremental_image – the name of incremental image to initialize, if specified will override default image

static is_fdw_schema(schema, unique_databases) → bool

load_data_into_relation(relation: snowshu.core.models.relation.Relation) → None

name = 'postgres'

static quoted(val: str) → str

Returns quoted value if appropriate.

replace_x00_values(relation: snowshu.core.models.relation.Relation) → snowshu.core.models.relation.Relation