Welcome!
prefect-sqlalchemy
helps you connect to a database in your Prefect flows.
Getting started
Install prefect-sqlalchemy
The following command will install a version of prefect-sqlalchemy
compatible with your installed version of prefect
.
If you don’t already have prefect
installed, it will install the newest version of prefect
as well.
pip install "prefect[sqlalchemy]"
Upgrade to the latest versions of prefect
and prefect-sqlalchemy
:
pip install -U "prefect[sqlalchemy]"
Register newly installed block types
Register the block types in the prefect-sqlalchemy
module to make them available for use.
prefect block register -m prefect_sqlalchemy
Examples
Save credentials to a block
To use the load
method on Blocks, you must have a block saved through code or saved through the UI.
from prefect_sqlalchemy import SqlAlchemyConnector, ConnectionComponents, SyncDriver
connector = SqlAlchemyConnector(
connection_info=ConnectionComponents(
driver=SyncDriver.POSTGRESQL_PSYCOPG2,
username="USERNAME-PLACEHOLDER",
password="PASSWORD-PLACEHOLDER",
host="localhost",
port=5432,
database="DATABASE-PLACEHOLDER",
)
)
connector.save("BLOCK_NAME-PLACEHOLDER")
Load the saved block that holds your credentials:
from prefect_sqlalchemy import SqlAlchemyConnector
SqlAlchemyConnector.load("BLOCK_NAME-PLACEHOLDER")
The required arguments depend upon the desired driver. For example, SQLite requires only the driver
and database
arguments:
from prefect_sqlalchemy import SqlAlchemyConnector, ConnectionComponents, SyncDriver
connector = SqlAlchemyConnector(
connection_info=ConnectionComponents(
driver=SyncDriver.SQLITE_PYSQLITE,
database="DATABASE-PLACEHOLDER.db"
)
)
connector.save("BLOCK_NAME-PLACEHOLDER")
Work with databases in a flow
To set up a table, use the execute
and execute_many
methods.
Use the fetch_many
method to retrieve data in a stream until there’s no more data.
Use the SqlAlchemyConnector
as a context manager, to ensure that the SQLAlchemy engine and any connected resources are closed properly after you’re done with them.
Async support
SqlAlchemyConnector
supports async workflows. Just be sure to save, load, and use an async driver, as in the example below.
from prefect_sqlalchemy import SqlAlchemyConnector, ConnectionComponents, AsyncDriver
connector = SqlAlchemyConnector(
connection_info=ConnectionComponents(
driver=AsyncDriver.SQLITE_AIOSQLITE,
database="DATABASE-PLACEHOLDER.db"
)
)
if __name__ == "__main__":
connector.save("BLOCK_NAME-PLACEHOLDER")
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_rows
if __name__ == "__main__":
sqlalchemy_flow("BLOCK-NAME-PLACEHOLDER")
Resources
Refer to the prefect-sqlalchemy
SDK documentation linked in the sidebar to explore all the capabilities of the prefect-sqlalchemy
library.
For assistance using SQLAlchemy, consult the SQLAlchemy documentation.