PostgreSQL

Add credentials

Create a new pipeline or open an existing pipeline.
Expand the left side of your screen to view the file browser.
Scroll down and click on a file named io_config.yaml.
Enter the following keys and values under the key named default (you can have multiple profiles, add it under whichever is relevant to you)

version: 0.1.1
default:
  POSTGRES_DBNAME: ...
  POSTGRES_HOST: ...
  POSTGRES_PASSWORD: ...
  POSTGRES_PORT: ...
  POSTGRES_USER: ...
  POSTGRES_CONNECTION_METHOD: direct
  POSTGRES_SSH_HOST:
  POSTGRES_SSH_PORT: 22
  POSTGRES_SSH_USERNAME:
  POSTGRES_SSH_PASSWORD:
  POSTGRES_SSH_PKEY:

Pro-only fields

  POSTGRES_SSL_MODE:
  POSTGRES_SSL_ROOTCERT:
  POSTGRES_SSL_CERT:
  POSTGRES_SSL_KEY:

Notes:

POSTGRES_SSL_MODE corresponds to the sslmode parameter in psycopg2 (e.g., require, verify-full).
POSTGRES_SSL_ROOTCERT, POSTGRES_SSL_CERT, and POSTGRES_SSL_KEY should be absolute paths to your SSL files on disk.

SSH tunneling

If you want to connect to Postgres cluster with ssh tunnel, update the value of POSTGRES_CONNECTION_METHOD to ssh_tunnel, and enter the values for keys with prefix POSTGRES_SSH.

Key	Description	Sample value
`POSTGRES_SSH_HOST`	The host of the intermediate bastion server.	`123.45.67.89`
`POSTGRES_SSH_PORT`	The port of the intermediate bastion server. Default value: 22	`22`
`POSTGRES_SSH_USERNAME`	The username used to connect to the bastion server.	`username`
`POSTGRES_SSH_PASSWORD`	(Optional) The password used to connect to the bastion server. It should be set if you authenticate with the bastion server with password.	`password`
`POSTGRES_SSH_PKEY`	(Optional) The path to the private key used to connect to the bastion server. It should be set if you authenticate with the bastion server with private key.	`/path/to/private/key`

Using SQL block

Create a new pipeline or open an existing pipeline.
Add a data loader, transformer, or data exporter block.
Select SQL.
Under the Data provider dropdown, select PostgreSQL.
Under the Profile dropdown, select default (or the profile you added credentials underneath).
Next to the Save to schema label, enter the schema name you want this block to save data to.
Under the Write policy dropdown, select Replace or Append (please see SQL blocks guide for more information on write policies).
Enter in this test query: SELECT 1.
Run the block.

Using Python block

Create a new pipeline or open an existing pipeline.
Add a data loader, transformer, or data exporter block (the code snippet below is for a data loader).
Select Generic (no template).
Enter this code snippet (note: change the config_profile from default if you have a different profile):

from mage_ai.settings.repo import get_repo_path
from mage_ai.io.config import ConfigFileLoader
from mage_ai.io.postgres import Postgres
from os import path
from pandas import DataFrame

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader


@data_loader
def load_data_from_postgres(**kwargs) -> DataFrame:
    query = 'SELECT 1'
    config_path = path.join(get_repo_path(), 'io_config.yaml')
    config_profile = 'default'

    with Postgres.with_config(ConfigFileLoader(config_path, config_profile)) as loader:
        return loader.load(query)

Run the block.

Export a dataframe

Here is an example code snippet to export a dataframe to Postgres:

from mage_ai.settings.repo import get_repo_path
from mage_ai.io.config import ConfigFileLoader
from mage_ai.io.postgres import Postgres
from pandas import DataFrame
from os import path

if 'data_exporter' not in globals():
    from mage_ai.data_preparation.decorators import data_exporter


@data_exporter
def export_data_to_postgres(df: DataFrame, **kwargs) -> None:
    schema_name = 'your_schema_name'  # Specify the name of the schema to export data to
    table_name = 'your_table_name'  # Specify the name of the table to export data to
    config_path = path.join(get_repo_path(), 'io_config.yaml')
    config_profile = 'default'

    with Postgres.with_config(ConfigFileLoader(config_path, config_profile)) as loader:
        loader.export(
            df,
            schema_name,
            table_name,
            index=False,  # Specifies whether to include index in exported table
            if_exists='replace',  # Specify resolution policy if table name already exists
            drop_table_on_replace=False,   # Whether to drop the table when "if_exists" param is set to "replace"
            allow_reserved_words=True,
            unique_conflict_method='UPDATE',
            unique_constraints=['col'],
        )

Custom types

To overwrite a column type when running a python export block, simply specify the column name and type in the overwrite_types dict in data exporter config Here is an example code snippet:

from mage_ai.settings.repo import get_repo_path
from mage_ai.io.config import ConfigFileLoader
from mage_ai.io.postgres import Postgres
from pandas import DataFrame
from os import path

if 'data_exporter' not in globals():
    from mage_ai.data_preparation.decorators import data_exporter


@data_exporter
def export_data_to_postgres(df: DataFrame, **kwargs) -> None:
    schema_name = 'your_schema_name'  # Specify the name of the schema to export data to
    table_name = 'your_table_name'  # Specify the name of the table to export data to
    config_path = path.join(get_repo_path(), 'io_config.yaml')
    config_profile = 'default'
    overwrite_types = {'column_name': 'VARCHAR(255)'}

    with Postgres.with_config(ConfigFileLoader(config_path, config_profile)) as loader:
        loader.export(
            df,
            schema_name,
            table_name,
            index=False,  # Specifies whether to include index in exported table
            if_exists='replace',  # Specify resolution policy if table name already exists
            allow_reserved_words=True,
            unique_conflict_method='UPDATE',
            unique_constraints=['col'],
            overwrite_types=overwrite_types,
        )

Method arguments

Field name	Description	Example values
allow_reserved_words	Whether to allow using reserved words as column names.	True/False (default: False)
auto_clean_name	Whether to automatically clean the column name (replace the empty space with underscore, avoid using number as the prefix of the column name)	True/False (default: True)
case_sensitive	Whether to support case sensitive columns	True/False (default: False)
cascade_on_drop	Whether to cascade the table drop.	True/False (default: False)
drop_table_on_replace	Whether to drop the table when “if_exists” param is set to “replace”.	True/False (default: False)
if_exists	Specify resolution policy if table name already exists	”fail”/“replace”/“append” (default: “replace”)
overwrite_types	Overwrite the column types	`{'column1': 'INTEGER', 'column2': 'VARCHAR'}`
unique_conflict_method	How to handle the conflict on unique constraints.	‘UPDATE’ (default: None)
unique_constraints	The unique constraints of the table.	[‘col1’, ‘col2’] (default: None)

Pro-only fields

Field name	Description	Example values
batch_write	Use `COPY` command to write to PostgreSQL to improve performance.	True/False (default: False)

Infrastructure

3rd party

Add credentials

Pro-only fields

SSH tunneling

Using SQL block

Using Python block

Export a dataframe

Method arguments

Pro-only fields

​Add credentials

​Pro-only fields

​SSH tunneling

​Using SQL block

​Using Python block

​Export a dataframe

​Method arguments

​Pro-only fields

Add credentials

Pro-only fields

SSH tunneling

Using SQL block

Using Python block

Export a dataframe

Method arguments

Pro-only fields