Storage
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:
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, selectPostgreSQL
. - Under the
Profile
dropdown, selectdefault
(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, selectReplace
orAppend
(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
fromdefault
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
allow_reserved_words=True,
unique_conflict_method='UPDATE',
unique_constraints=['col'],
)