bash
docker run -d --name prefect-postgres \
    -v prefectdb:/var/lib/postgresql/data \
    -p 5432:5432 \
    -e POSTGRES_USER=postgres \
    -e POSTGRES_PASSWORD=yourTopSecretPassword \
    -e POSTGRES_DB=prefect postgres:latest

SqlAlchemyConnector

SqlAlchemyConnector

python
from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector

@task
def setup_table(block_name: str) -> None:
    with SqlAlchemyConnector.load(block_name) as connector:
        connector.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
        connector.execute("INSERT INTO customers (name, address) VALUES (:name, :address);", parameters={"name": "Marvin", "address": "Highway 42"})
        connector.execute_many("INSERT INTO customers (name, address) VALUES (:name, :address);", seq_of_parameters=[{"name": "Ford", "address": "Highway 42"}, {"name": "Unknown", "address": "Highway 42"}])

@task
def fetch_data(block_name: str) -> list:
    all_rows = []
    with SqlAlchemyConnector.load(block_name) as connector:
        while True:
            new_rows = connector.fetch_many("SELECT * FROM customers", size=2)
            if len(new_rows) == 0:
                break
            all_rows.append(new_rows)
    return all_rows

@flow
def sqlalchemy_flow(block_name: str) -> list:
    setup_table(block_name)
    all_rows = fetch_data(block_name)
    return all

if __name__ == "__main__":
    sqlalchemy_flow("YOUR_BLOCK_NAME")

YOUR_BLOCK_NAME

bash
   docker run -d --name prefect-postgres \
       -v prefectdb:/var/lib/postgresql/data \
       -p 5432:5432 \
       -e POSTGRES_USER=postgres \
       -e POSTGRES_PASSWORD=yourTopSecretPassword \
       -e POSTGRES_DB=prefect postgres:latest

bash
   prefect config set PREFECT_API_DATABASE_CONNECTION_URL="<postgresql+asyncpg://postgres:yourTopSecretPassword@localhost:5432/prefect>"

<postgres+asyncpg://postgres:yourTopSecretPassword@localhost:5432/your_database_name>

my-postgres-block

python
from prefect import flow, task
from prefect_sqlalchemy import SqlAlchemyConnector

@task
def setup_source_table(block_name: str) -> None:
    with SqlAlchemyConnector.load(block_name) as connector:
        connector.execute("CREATE TABLE IF NOT EXISTS source_table (id SERIAL PRIMARY KEY, name VARCHAR);")
        connector.execute("INSERT INTO source_table (name) VALUES ('item1'), ('item2'), ('item3');")

@task
def setup_destination_table(block_name: str) -> None:
    with SqlAlchemyConnector.load(block_name) as connector:
        connector.execute("CREATE TABLE IF NOT EXISTS destination_table (id SERIAL PRIMARY KEY, name VARCHAR);")

@task
def transfer_data(block_name: str) -> list:
    with SqlAlchemyConnector.load(block_name) as connector:
        data = connector.fetch_all("SELECT * FROM source_table;")
        for row in data:
            connector.execute("INSERT INTO destination_table (name) VALUES (:name);", parameters={"name": row['name']})
    return data

@flow
def postgres_pipeline(block_name: str):
    setup_source_table(block_name)
    setup_destination_table(block_name)
    transferred_data = transfer_data(block_name)
    print(f"Data transferred: {transferred_data}")

if __name__ == "__main__":
    postgres_pipeline("my-postgres-block")

bash
   python your_script_name.py

python
   from prefect.deployments import Deployment

   deployment = Deployment.build_from_flow(
       flow=postgres_pipeline,
       name='postgres-pipeline-deployment'
   )

   deployment.apply()