Skip to content

prefect_sqlalchemy.database

Tasks for querying a database with SQLAlchemy

SqlAlchemyConnector

Bases: CredentialsBlock, DatabaseBlock

Block used to manage authentication with a database.

Upon instantiating, an engine is created and maintained for the life of the object until the close method is called.

It is recommended to use this block as a context manager, which will automatically close the engine and its connections when the context is exited.

It is also recommended that this block is loaded and consumed within a single task or flow because if the block is passed across separate tasks and flows, the state of the block's connection and cursor could be lost.

Attributes:

Name Type Description
connection_info Union[ConnectionComponents, AnyUrl]

SQLAlchemy URL to create the engine; either create from components or create from a string.

connect_args Optional[Dict[str, Any]]

The options which will be passed directly to the DBAPI's connect() method as additional keyword arguments.

fetch_size int

The number of rows to fetch at a time.

Example

Load stored database credentials and use in context manager:

from prefect_sqlalchemy import SqlAlchemyConnector

database_block = SqlAlchemyConnector.load("BLOCK_NAME")
with database_block:
    ...

Create table named customers and insert values; then fetch the first 10 rows.

from prefect_sqlalchemy import (
    SqlAlchemyConnector, SyncDriver, ConnectionComponents
)

with SqlAlchemyConnector(
    connection_info=ConnectionComponents(
        driver=SyncDriver.SQLITE_PYSQLITE,
        database="prefect.db"
    )
) as database:
    database.execute(
        "CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);",
    )
    for i in range(1, 42):
        database.execute(
            "INSERT INTO customers (name, address) VALUES (:name, :address);",
            parameters={"name": "Marvin", "address": f"Highway {i}"},
        )
    results = database.fetch_many(
        "SELECT * FROM customers WHERE name = :name;",
        parameters={"name": "Marvin"},
        size=10
    )
print(results)

Source code in prefect_sqlalchemy/database.py
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
class SqlAlchemyConnector(CredentialsBlock, DatabaseBlock):
    """
    Block used to manage authentication with a database.

    Upon instantiating, an engine is created and maintained for the life of
    the object until the close method is called.

    It is recommended to use this block as a context manager, which will automatically
    close the engine and its connections when the context is exited.

    It is also recommended that this block is loaded and consumed within a single task
    or flow because if the block is passed across separate tasks and flows,
    the state of the block's connection and cursor could be lost.

    Attributes:
        connection_info: SQLAlchemy URL to create the engine;
            either create from components or create from a string.
        connect_args: The options which will be passed directly to the
            DBAPI's connect() method as additional keyword arguments.
        fetch_size: The number of rows to fetch at a time.

    Example:
        Load stored database credentials and use in context manager:
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        database_block = SqlAlchemyConnector.load("BLOCK_NAME")
        with database_block:
            ...
        ```

        Create table named customers and insert values; then fetch the first 10 rows.
        ```python
        from prefect_sqlalchemy import (
            SqlAlchemyConnector, SyncDriver, ConnectionComponents
        )

        with SqlAlchemyConnector(
            connection_info=ConnectionComponents(
                driver=SyncDriver.SQLITE_PYSQLITE,
                database="prefect.db"
            )
        ) as database:
            database.execute(
                "CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);",
            )
            for i in range(1, 42):
                database.execute(
                    "INSERT INTO customers (name, address) VALUES (:name, :address);",
                    parameters={"name": "Marvin", "address": f"Highway {i}"},
                )
            results = database.fetch_many(
                "SELECT * FROM customers WHERE name = :name;",
                parameters={"name": "Marvin"},
                size=10
            )
        print(results)
        ```
    """

    _block_type_name = "SQLAlchemy Connector"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/3c7dff04f70aaf4528e184a3b028f9e40b98d68c-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-sqlalchemy/database/#prefect_sqlalchemy.database.SqlAlchemyConnector"  # noqa

    connection_info: Union[ConnectionComponents, AnyUrl] = Field(
        default=...,
        description=(
            "SQLAlchemy URL to create the engine; either create from components "
            "or create from a string."
        ),
    )
    connect_args: Optional[Dict[str, Any]] = Field(
        default=None,
        title="Additional Connection Arguments",
        description=(
            "The options which will be passed directly to the DBAPI's connect() "
            "method as additional keyword arguments."
        ),
    )
    fetch_size: int = Field(
        default=1, description="The number of rows to fetch at a time."
    )

    _engine: Optional[Union[AsyncEngine, Engine]] = None
    _exit_stack: Union[ExitStack, AsyncExitStack] = None
    _unique_results: Dict[str, CursorResult] = None

    class Config:
        """Configuration of pydantic."""

        # Support serialization of the 'URL' type
        arbitrary_types_allowed = True
        json_encoders = {URL: lambda u: u.render_as_string()}

    def dict(self, *args, **kwargs) -> Dict:
        """
        Convert to a dictionary.
        """
        # Support serialization of the 'URL' type
        d = super().dict(*args, **kwargs)
        d["_rendered_url"] = SecretStr(
            self._rendered_url.render_as_string(hide_password=False)
        )
        return d

    def block_initialization(self):
        """
        Initializes the engine.
        """
        super().block_initialization()

        if isinstance(self.connection_info, ConnectionComponents):
            self._rendered_url = self.connection_info.create_url()
        else:
            # make rendered url from string
            self._rendered_url = make_url(str(self.connection_info))
        drivername = self._rendered_url.drivername

        try:
            AsyncDriver(drivername)
            self._driver_is_async = True
        except ValueError:
            self._driver_is_async = False

        if self._unique_results is None:
            self._unique_results = {}

        if self._exit_stack is None:
            self._start_exit_stack()

    def _start_exit_stack(self):
        """
        Starts an AsyncExitStack or ExitStack depending on whether driver is async.
        """
        self._exit_stack = AsyncExitStack() if self._driver_is_async else ExitStack()

    def get_engine(
        self, **create_engine_kwargs: Dict[str, Any]
    ) -> Union[Engine, AsyncEngine]:
        """
        Returns an authenticated engine that can be
        used to query from databases.

        If an existing engine exists, return that one.

        Returns:
            The authenticated SQLAlchemy Engine / AsyncEngine.

        Examples:
            Create an asynchronous engine to PostgreSQL using URL params.
            ```python
            from prefect import flow
            from prefect_sqlalchemy import (
                SqlAlchemyConnector, ConnectionComponents, AsyncDriver
            )

            @flow
            def sqlalchemy_credentials_flow():
                sqlalchemy_credentials = SqlAlchemyConnector(
                connection_info=ConnectionComponents(
                        driver=AsyncDriver.POSTGRESQL_ASYNCPG,
                        username="prefect",
                        password="prefect_password",
                        database="postgres"
                    )
                )
                print(sqlalchemy_credentials.get_engine())

            sqlalchemy_credentials_flow()
            ```

            Create a synchronous engine to Snowflake using the `url` kwarg.
            ```python
            from prefect import flow
            from prefect_sqlalchemy import SqlAlchemyConnector, AsyncDriver

            @flow
            def sqlalchemy_credentials_flow():
                url = (
                    "snowflake://<user_login_name>:<password>"
                    "@<account_identifier>/<database_name>"
                    "?warehouse=<warehouse_name>"
                )
                sqlalchemy_credentials = SqlAlchemyConnector(url=url)
                print(sqlalchemy_credentials.get_engine())

            sqlalchemy_credentials_flow()
            ```
        """
        if self._engine is not None:
            self.logger.debug("Reusing existing engine.")
            return self._engine

        engine_kwargs = dict(
            url=self._rendered_url,
            connect_args=self.connect_args or {},
            **create_engine_kwargs,
        )
        if self._driver_is_async:
            # no need to await here
            engine = create_async_engine(**engine_kwargs)
        else:
            engine = create_engine(**engine_kwargs)
        self.logger.info("Created a new engine.")

        if self._engine is None:
            self._engine = engine

        return engine

    def get_connection(
        self, begin: bool = True, **connect_kwargs: Dict[str, Any]
    ) -> Union[Connection, AsyncConnection]:
        """
        Returns a connection that can be used to query from databases.

        Args:
            begin: Whether to begin a transaction on the connection; if True, if
                any operations fail, the entire transaction will be rolled back.
            **connect_kwargs: Additional keyword arguments to pass to either
                `engine.begin` or engine.connect`.

        Returns:
            The SQLAlchemy Connection / AsyncConnection.

        Examples:
            Create an synchronous connection as a context-managed transaction.
            ```python
            from prefect_sqlalchemy import SqlAlchemyConnector

            sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
            with sqlalchemy_connector.get_connection(begin=False) as connection:
                connection.execute("SELECT * FROM table LIMIT 1;")
            ```

            Create an asynchronous connection as a context-managed transacation.
            ```python
            import asyncio
            from prefect_sqlalchemy import SqlAlchemyConnector

            sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
            async with sqlalchemy_connector.get_connection(begin=False) as connection:
                asyncio.run(connection.execute("SELECT * FROM table LIMIT 1;"))
            ```
        """  # noqa: E501
        engine = self.get_engine()
        if begin:
            connection = engine.begin(**connect_kwargs)
        else:
            connection = engine.connect(**connect_kwargs)
        self.logger.info("Created a new connection.")
        return connection

    def get_client(
        self,
        client_type: Literal["engine", "connection"],
        **get_client_kwargs: Dict[str, Any],
    ) -> Union[Engine, AsyncEngine, Connection, AsyncConnection]:
        """
        Returns either an engine or connection that can be used to query from databases.

        Args:
            client_type: Select from either 'engine' or 'connection'.
            **get_client_kwargs: Additional keyword arguments to pass to
                either `get_engine` or `get_connection`.

        Returns:
            The authenticated SQLAlchemy engine or connection.

        Examples:
            Create an engine.
            ```python
            from prefect_sqlalchemy import SqlalchemyConnector

            sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
            engine = sqlalchemy_connector.get_client(client_type="engine")
            ```

            Create a context managed connection.
            ```python
            from prefect_sqlalchemy import SqlalchemyConnector

            sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
            with sqlalchemy_connector.get_client(client_type="connection") as conn:
                ...
            ```
        """  # noqa: E501
        if client_type == "engine":
            client = self.get_engine(**get_client_kwargs)
        elif client_type == "connection":
            client = self.get_connection(**get_client_kwargs)
        else:
            raise ValueError(
                f"{client_type!r} is not supported; choose from engine or connection."
            )
        return client

    async def _async_sync_execute(
        self,
        connection: Union[Connection, AsyncConnection],
        *execute_args: Tuple[Any],
        **execute_kwargs: Dict[str, Any],
    ) -> CursorResult:
        """
        Execute the statement asynchronously or synchronously.
        """
        # can't use run_sync_in_worker_thread:
        # ProgrammingError: (sqlite3.ProgrammingError) SQLite objects created in a
        # thread can only be used in that same thread.
        result_set = connection.execute(*execute_args, **execute_kwargs)

        if self._driver_is_async:
            result_set = await result_set
            await connection.commit()  # very important
        elif SQLALCHEMY_VERSION.startswith("2."):
            connection.commit()
        return result_set

    @asynccontextmanager
    async def _manage_connection(self, **get_connection_kwargs: Dict[str, Any]):
        if self._driver_is_async:
            async with self.get_connection(**get_connection_kwargs) as connection:
                yield connection
        else:
            with self.get_connection(**get_connection_kwargs) as connection:
                yield connection

    async def _get_result_set(
        self, *execute_args: Tuple[Any], **execute_kwargs: Dict[str, Any]
    ) -> CursorResult:
        """
        Returns a new or existing result set based on whether the inputs
        are unique.

        Args:
            *execute_args: Args to pass to execute.
            **execute_kwargs: Keyword args to pass to execute.

        Returns:
            The result set from the operation.
        """  # noqa: E501
        input_hash = hash_objects(*execute_args, **execute_kwargs)
        assert input_hash is not None, (
            "We were not able to hash your inputs, "
            "which resulted in an unexpected data return; "
            "please open an issue with a reproducible example."
        )

        if input_hash not in self._unique_results.keys():
            if self._driver_is_async:
                connection = await self._exit_stack.enter_async_context(
                    self.get_connection()
                )
            else:
                connection = self._exit_stack.enter_context(self.get_connection())
            result_set = await self._async_sync_execute(
                connection, *execute_args, **execute_kwargs
            )
            # implicitly store the connection by storing the result set
            # which points to its parent connection
            self._unique_results[input_hash] = result_set
        else:
            result_set = self._unique_results[input_hash]
        return result_set

    def _reset_cursor_results(self) -> None:
        """
        Closes all the existing cursor results.
        """
        input_hashes = tuple(self._unique_results.keys())
        for input_hash in input_hashes:
            try:
                cursor_result = self._unique_results.pop(input_hash)
                cursor_result.close()
            except Exception as exc:
                self.logger.warning(
                    f"Failed to close connection for input hash {input_hash!r}: {exc}"
                )

    @sync_compatible
    async def reset_connections(self) -> None:
        """
        Tries to close all opened connections and their results.

        Examples:
            Resets connections so `fetch_*` methods return new results.
            ```python
            from prefect_sqlalchemy import SqlAlchemyConnector

            with SqlAlchemyConnector.load("MY_BLOCK") as database:
                results = database.fetch_one("SELECT * FROM customers")
                database.reset_connections()
                results = database.fetch_one("SELECT * FROM customers")
            ```
        """
        if self._driver_is_async:
            raise RuntimeError(
                f"{self._rendered_url.drivername} has no synchronous connections. "
                f"Please use the `reset_async_connections` method instead."
            )

        if self._exit_stack is None:
            self.logger.info("There were no connections to reset.")
            return

        self._reset_cursor_results()
        self._exit_stack.close()
        self.logger.info("Reset opened connections and their results.")

    async def reset_async_connections(self) -> None:
        """
        Tries to close all opened connections and their results.

        Examples:
            Resets connections so `fetch_*` methods return new results.
            ```python
            import asyncio
            from prefect_sqlalchemy import SqlAlchemyConnector

            async def example_run():
                async with SqlAlchemyConnector.load("MY_BLOCK") as database:
                    results = await database.fetch_one("SELECT * FROM customers")
                    await database.reset_async_connections()
                    results = await database.fetch_one("SELECT * FROM customers")

            asyncio.run(example_run())
            ```
        """
        if not self._driver_is_async:
            raise RuntimeError(
                f"{self._rendered_url.drivername} has no asynchronous connections. "
                f"Please use the `reset_connections` method instead."
            )

        if self._exit_stack is None:
            self.logger.info("There were no connections to reset.")
            return

        self._reset_cursor_results()
        await self._exit_stack.aclose()
        self.logger.info("Reset opened connections and their results.")

    @sync_compatible
    async def fetch_one(
        self,
        operation: str,
        parameters: Optional[Dict[str, Any]] = None,
        **execution_options: Dict[str, Any],
    ) -> Tuple[Any]:
        """
        Fetch a single result from the database.

        Repeated calls using the same inputs to *any* of the fetch methods of this
        block will skip executing the operation again, and instead,
        return the next set of results from the previous execution,
        until the reset_cursors method is called.

        Args:
            operation: The SQL query or other operation to be executed.
            parameters: The parameters for the operation.
            **execution_options: Options to pass to `Connection.execution_options`.

        Returns:
            A list of tuples containing the data returned by the database,
                where each row is a tuple and each column is a value in the tuple.

        Examples:
            Create a table, insert three rows into it, and fetch a row repeatedly.
            ```python
            from prefect_sqlalchemy import SqlAlchemyConnector

            with SqlAlchemyConnector.load("MY_BLOCK") as database:
                database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
                database.execute_many(
                    "INSERT INTO customers (name, address) VALUES (:name, :address);",
                    seq_of_parameters=[
                        {"name": "Ford", "address": "Highway 42"},
                        {"name": "Unknown", "address": "Space"},
                        {"name": "Me", "address": "Myway 88"},
                    ],
                )
                results = True
                while results:
                    results = database.fetch_one("SELECT * FROM customers")
                    print(results)
            ```
        """  # noqa
        result_set = await self._get_result_set(
            text(operation), parameters, execution_options=execution_options
        )
        self.logger.debug("Preparing to fetch one row.")
        row = result_set.fetchone()
        return row

    @sync_compatible
    async def fetch_many(
        self,
        operation: str,
        parameters: Optional[Dict[str, Any]] = None,
        size: Optional[int] = None,
        **execution_options: Dict[str, Any],
    ) -> List[Tuple[Any]]:
        """
        Fetch a limited number of results from the database.

        Repeated calls using the same inputs to *any* of the fetch methods of this
        block will skip executing the operation again, and instead,
        return the next set of results from the previous execution,
        until the reset_cursors method is called.

        Args:
            operation: The SQL query or other operation to be executed.
            parameters: The parameters for the operation.
            size: The number of results to return; if None or 0, uses the value of
                `fetch_size` configured on the block.
            **execution_options: Options to pass to `Connection.execution_options`.

        Returns:
            A list of tuples containing the data returned by the database,
                where each row is a tuple and each column is a value in the tuple.

        Examples:
            Create a table, insert three rows into it, and fetch two rows repeatedly.
            ```python
            from prefect_sqlalchemy import SqlAlchemyConnector

            with SqlAlchemyConnector.load("MY_BLOCK") as database:
                database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
                database.execute_many(
                    "INSERT INTO customers (name, address) VALUES (:name, :address);",
                    seq_of_parameters=[
                        {"name": "Ford", "address": "Highway 42"},
                        {"name": "Unknown", "address": "Space"},
                        {"name": "Me", "address": "Myway 88"},
                    ],
                )
                results = database.fetch_many("SELECT * FROM customers", size=2)
                print(results)
                results = database.fetch_many("SELECT * FROM customers", size=2)
                print(results)
            ```
        """  # noqa
        result_set = await self._get_result_set(
            text(operation), parameters, execution_options=execution_options
        )
        size = size or self.fetch_size
        self.logger.debug(f"Preparing to fetch {size} rows.")
        rows = result_set.fetchmany(size=size)
        return rows

    @sync_compatible
    async def fetch_all(
        self,
        operation: str,
        parameters: Optional[Dict[str, Any]] = None,
        **execution_options: Dict[str, Any],
    ) -> List[Tuple[Any]]:
        """
        Fetch all results from the database.

        Repeated calls using the same inputs to *any* of the fetch methods of this
        block will skip executing the operation again, and instead,
        return the next set of results from the previous execution,
        until the reset_cursors method is called.

        Args:
            operation: The SQL query or other operation to be executed.
            parameters: The parameters for the operation.
            **execution_options: Options to pass to `Connection.execution_options`.

        Returns:
            A list of tuples containing the data returned by the database,
                where each row is a tuple and each column is a value in the tuple.

        Examples:
            Create a table, insert three rows into it, and fetch all where name is 'Me'.
            ```python
            from prefect_sqlalchemy import SqlAlchemyConnector

            with SqlAlchemyConnector.load("MY_BLOCK") as database:
                database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
                database.execute_many(
                    "INSERT INTO customers (name, address) VALUES (:name, :address);",
                    seq_of_parameters=[
                        {"name": "Ford", "address": "Highway 42"},
                        {"name": "Unknown", "address": "Space"},
                        {"name": "Me", "address": "Myway 88"},
                    ],
                )
                results = database.fetch_all("SELECT * FROM customers WHERE name = :name", parameters={"name": "Me"})
            ```
        """  # noqa
        result_set = await self._get_result_set(
            text(operation), parameters, execution_options=execution_options
        )
        self.logger.debug("Preparing to fetch all rows.")
        rows = result_set.fetchall()
        return rows

    @sync_compatible
    async def execute(
        self,
        operation: str,
        parameters: Optional[Dict[str, Any]] = None,
        **execution_options: Dict[str, Any],
    ) -> None:
        """
        Executes an operation on the database. This method is intended to be used
        for operations that do not return data, such as INSERT, UPDATE, or DELETE.

        Unlike the fetch methods, this method will always execute the operation
        upon calling.

        Args:
            operation: The SQL query or other operation to be executed.
            parameters: The parameters for the operation.
            **execution_options: Options to pass to `Connection.execution_options`.

        Examples:
            Create a table and insert one row into it.
            ```python
            from prefect_sqlalchemy import SqlAlchemyConnector

            with SqlAlchemyConnector.load("MY_BLOCK") as database:
                database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
                database.execute(
                    "INSERT INTO customers (name, address) VALUES (:name, :address);",
                    parameters={"name": "Marvin", "address": "Highway 42"},
                )
            ```
        """  # noqa
        async with self._manage_connection(begin=False) as connection:
            await self._async_sync_execute(
                connection,
                text(operation),
                parameters,
                execution_options=execution_options,
            )
        self.logger.info(f"Executed the operation, {operation!r}")

    @sync_compatible
    async def execute_many(
        self,
        operation: str,
        seq_of_parameters: List[Dict[str, Any]],
        **execution_options: Dict[str, Any],
    ) -> None:
        """
        Executes many operations on the database. This method is intended to be used
        for operations that do not return data, such as INSERT, UPDATE, or DELETE.

        Unlike the fetch methods, this method will always execute the operation
        upon calling.

        Args:
            operation: The SQL query or other operation to be executed.
            seq_of_parameters: The sequence of parameters for the operation.
            **execution_options: Options to pass to `Connection.execution_options`.

        Examples:
            Create a table and insert two rows into it.
            ```python
            from prefect_sqlalchemy import SqlAlchemyConnector

            with SqlAlchemyConnector.load("MY_BLOCK") as database:
                database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
                database.execute_many(
                    "INSERT INTO customers (name, address) VALUES (:name, :address);",
                    seq_of_parameters=[
                        {"name": "Ford", "address": "Highway 42"},
                        {"name": "Unknown", "address": "Space"},
                        {"name": "Me", "address": "Myway 88"},
                    ],
                )
            ```
        """  # noqa
        async with self._manage_connection(begin=False) as connection:
            await self._async_sync_execute(
                connection,
                text(operation),
                seq_of_parameters,
                execution_options=execution_options,
            )
        self.logger.info(
            f"Executed {len(seq_of_parameters)} operations based off {operation!r}."
        )

    async def __aenter__(self):
        """
        Start an asynchronous database engine upon entry.
        """
        if not self._driver_is_async:
            raise RuntimeError(
                f"{self._rendered_url.drivername} cannot be run asynchronously. "
                f"Please use the `with` syntax."
            )
        return self

    async def __aexit__(self, *args):
        """
        Dispose the asynchronous database engine upon exit.
        """
        await self.aclose()

    async def aclose(self):
        """
        Closes async connections and its cursors.
        """
        if not self._driver_is_async:
            raise RuntimeError(
                f"{self._rendered_url.drivername} is not asynchronous. "
                f"Please use the `close` method instead."
            )
        try:
            await self.reset_async_connections()
        finally:
            if self._engine is not None:
                await self._engine.dispose()
                self._engine = None
                self.logger.info("Disposed the engine.")

    def __enter__(self):
        """
        Start an synchronous database engine upon entry.
        """
        if self._driver_is_async:
            raise RuntimeError(
                f"{self._rendered_url.drivername} cannot be run synchronously. "
                f"Please use the `async with` syntax."
            )
        return self

    def __exit__(self, *args):
        """
        Dispose the synchronous database engine upon exit.
        """
        self.close()

    def close(self):
        """
        Closes sync connections and its cursors.
        """
        if self._driver_is_async:
            raise RuntimeError(
                f"{self._rendered_url.drivername} is not synchronous. "
                f"Please use the `aclose` method instead."
            )

        try:
            self.reset_connections()
        finally:
            if self._engine is not None:
                self._engine.dispose()
                self._engine = None
                self.logger.info("Disposed the engine.")

    def __getstate__(self):
        """Allows the block to be pickleable."""
        data = self.__dict__.copy()
        data.update({k: None for k in {"_engine", "_exit_stack", "_unique_results"}})
        return data

    def __setstate__(self, data: dict):
        """Upon loading back, restart the engine and results."""
        self.__dict__.update(data)

        if self._unique_results is None:
            self._unique_results = {}

        if self._exit_stack is None:
            self._start_exit_stack()

Config

Configuration of pydantic.

Source code in prefect_sqlalchemy/database.py
277
278
279
280
281
282
class Config:
    """Configuration of pydantic."""

    # Support serialization of the 'URL' type
    arbitrary_types_allowed = True
    json_encoders = {URL: lambda u: u.render_as_string()}

aclose async

Closes async connections and its cursors.

Source code in prefect_sqlalchemy/database.py
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
async def aclose(self):
    """
    Closes async connections and its cursors.
    """
    if not self._driver_is_async:
        raise RuntimeError(
            f"{self._rendered_url.drivername} is not asynchronous. "
            f"Please use the `close` method instead."
        )
    try:
        await self.reset_async_connections()
    finally:
        if self._engine is not None:
            await self._engine.dispose()
            self._engine = None
            self.logger.info("Disposed the engine.")

block_initialization

Initializes the engine.

Source code in prefect_sqlalchemy/database.py
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def block_initialization(self):
    """
    Initializes the engine.
    """
    super().block_initialization()

    if isinstance(self.connection_info, ConnectionComponents):
        self._rendered_url = self.connection_info.create_url()
    else:
        # make rendered url from string
        self._rendered_url = make_url(str(self.connection_info))
    drivername = self._rendered_url.drivername

    try:
        AsyncDriver(drivername)
        self._driver_is_async = True
    except ValueError:
        self._driver_is_async = False

    if self._unique_results is None:
        self._unique_results = {}

    if self._exit_stack is None:
        self._start_exit_stack()

close

Closes sync connections and its cursors.

Source code in prefect_sqlalchemy/database.py
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
def close(self):
    """
    Closes sync connections and its cursors.
    """
    if self._driver_is_async:
        raise RuntimeError(
            f"{self._rendered_url.drivername} is not synchronous. "
            f"Please use the `aclose` method instead."
        )

    try:
        self.reset_connections()
    finally:
        if self._engine is not None:
            self._engine.dispose()
            self._engine = None
            self.logger.info("Disposed the engine.")

dict

Convert to a dictionary.

Source code in prefect_sqlalchemy/database.py
284
285
286
287
288
289
290
291
292
293
def dict(self, *args, **kwargs) -> Dict:
    """
    Convert to a dictionary.
    """
    # Support serialization of the 'URL' type
    d = super().dict(*args, **kwargs)
    d["_rendered_url"] = SecretStr(
        self._rendered_url.render_as_string(hide_password=False)
    )
    return d

execute async

Executes an operation on the database. This method is intended to be used for operations that do not return data, such as INSERT, UPDATE, or DELETE.

Unlike the fetch methods, this method will always execute the operation upon calling.

Parameters:

Name Type Description Default
operation str

The SQL query or other operation to be executed.

required
parameters Optional[Dict[str, Any]]

The parameters for the operation.

None
**execution_options Dict[str, Any]

Options to pass to Connection.execution_options.

{}

Examples:

Create a table and insert one row into it.

from prefect_sqlalchemy import SqlAlchemyConnector

with SqlAlchemyConnector.load("MY_BLOCK") as database:
    database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
    database.execute(
        "INSERT INTO customers (name, address) VALUES (:name, :address);",
        parameters={"name": "Marvin", "address": "Highway 42"},
    )

Source code in prefect_sqlalchemy/database.py
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
@sync_compatible
async def execute(
    self,
    operation: str,
    parameters: Optional[Dict[str, Any]] = None,
    **execution_options: Dict[str, Any],
) -> None:
    """
    Executes an operation on the database. This method is intended to be used
    for operations that do not return data, such as INSERT, UPDATE, or DELETE.

    Unlike the fetch methods, this method will always execute the operation
    upon calling.

    Args:
        operation: The SQL query or other operation to be executed.
        parameters: The parameters for the operation.
        **execution_options: Options to pass to `Connection.execution_options`.

    Examples:
        Create a table and insert one row into it.
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        with SqlAlchemyConnector.load("MY_BLOCK") as database:
            database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
            database.execute(
                "INSERT INTO customers (name, address) VALUES (:name, :address);",
                parameters={"name": "Marvin", "address": "Highway 42"},
            )
        ```
    """  # noqa
    async with self._manage_connection(begin=False) as connection:
        await self._async_sync_execute(
            connection,
            text(operation),
            parameters,
            execution_options=execution_options,
        )
    self.logger.info(f"Executed the operation, {operation!r}")

execute_many async

Executes many operations on the database. This method is intended to be used for operations that do not return data, such as INSERT, UPDATE, or DELETE.

Unlike the fetch methods, this method will always execute the operation upon calling.

Parameters:

Name Type Description Default
operation str

The SQL query or other operation to be executed.

required
seq_of_parameters List[Dict[str, Any]]

The sequence of parameters for the operation.

required
**execution_options Dict[str, Any]

Options to pass to Connection.execution_options.

{}

Examples:

Create a table and insert two rows into it.

from prefect_sqlalchemy import SqlAlchemyConnector

with SqlAlchemyConnector.load("MY_BLOCK") as database:
    database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
    database.execute_many(
        "INSERT INTO customers (name, address) VALUES (:name, :address);",
        seq_of_parameters=[
            {"name": "Ford", "address": "Highway 42"},
            {"name": "Unknown", "address": "Space"},
            {"name": "Me", "address": "Myway 88"},
        ],
    )

Source code in prefect_sqlalchemy/database.py
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
@sync_compatible
async def execute_many(
    self,
    operation: str,
    seq_of_parameters: List[Dict[str, Any]],
    **execution_options: Dict[str, Any],
) -> None:
    """
    Executes many operations on the database. This method is intended to be used
    for operations that do not return data, such as INSERT, UPDATE, or DELETE.

    Unlike the fetch methods, this method will always execute the operation
    upon calling.

    Args:
        operation: The SQL query or other operation to be executed.
        seq_of_parameters: The sequence of parameters for the operation.
        **execution_options: Options to pass to `Connection.execution_options`.

    Examples:
        Create a table and insert two rows into it.
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        with SqlAlchemyConnector.load("MY_BLOCK") as database:
            database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
            database.execute_many(
                "INSERT INTO customers (name, address) VALUES (:name, :address);",
                seq_of_parameters=[
                    {"name": "Ford", "address": "Highway 42"},
                    {"name": "Unknown", "address": "Space"},
                    {"name": "Me", "address": "Myway 88"},
                ],
            )
        ```
    """  # noqa
    async with self._manage_connection(begin=False) as connection:
        await self._async_sync_execute(
            connection,
            text(operation),
            seq_of_parameters,
            execution_options=execution_options,
        )
    self.logger.info(
        f"Executed {len(seq_of_parameters)} operations based off {operation!r}."
    )

fetch_all async

Fetch all results from the database.

Repeated calls using the same inputs to any of the fetch methods of this block will skip executing the operation again, and instead, return the next set of results from the previous execution, until the reset_cursors method is called.

Parameters:

Name Type Description Default
operation str

The SQL query or other operation to be executed.

required
parameters Optional[Dict[str, Any]]

The parameters for the operation.

None
**execution_options Dict[str, Any]

Options to pass to Connection.execution_options.

{}

Returns:

Type Description
List[Tuple[Any]]

A list of tuples containing the data returned by the database, where each row is a tuple and each column is a value in the tuple.

Examples:

Create a table, insert three rows into it, and fetch all where name is 'Me'.

from prefect_sqlalchemy import SqlAlchemyConnector

with SqlAlchemyConnector.load("MY_BLOCK") as database:
    database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
    database.execute_many(
        "INSERT INTO customers (name, address) VALUES (:name, :address);",
        seq_of_parameters=[
            {"name": "Ford", "address": "Highway 42"},
            {"name": "Unknown", "address": "Space"},
            {"name": "Me", "address": "Myway 88"},
        ],
    )
    results = database.fetch_all("SELECT * FROM customers WHERE name = :name", parameters={"name": "Me"})

Source code in prefect_sqlalchemy/database.py
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
@sync_compatible
async def fetch_all(
    self,
    operation: str,
    parameters: Optional[Dict[str, Any]] = None,
    **execution_options: Dict[str, Any],
) -> List[Tuple[Any]]:
    """
    Fetch all results from the database.

    Repeated calls using the same inputs to *any* of the fetch methods of this
    block will skip executing the operation again, and instead,
    return the next set of results from the previous execution,
    until the reset_cursors method is called.

    Args:
        operation: The SQL query or other operation to be executed.
        parameters: The parameters for the operation.
        **execution_options: Options to pass to `Connection.execution_options`.

    Returns:
        A list of tuples containing the data returned by the database,
            where each row is a tuple and each column is a value in the tuple.

    Examples:
        Create a table, insert three rows into it, and fetch all where name is 'Me'.
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        with SqlAlchemyConnector.load("MY_BLOCK") as database:
            database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
            database.execute_many(
                "INSERT INTO customers (name, address) VALUES (:name, :address);",
                seq_of_parameters=[
                    {"name": "Ford", "address": "Highway 42"},
                    {"name": "Unknown", "address": "Space"},
                    {"name": "Me", "address": "Myway 88"},
                ],
            )
            results = database.fetch_all("SELECT * FROM customers WHERE name = :name", parameters={"name": "Me"})
        ```
    """  # noqa
    result_set = await self._get_result_set(
        text(operation), parameters, execution_options=execution_options
    )
    self.logger.debug("Preparing to fetch all rows.")
    rows = result_set.fetchall()
    return rows

fetch_many async

Fetch a limited number of results from the database.

Repeated calls using the same inputs to any of the fetch methods of this block will skip executing the operation again, and instead, return the next set of results from the previous execution, until the reset_cursors method is called.

Parameters:

Name Type Description Default
operation str

The SQL query or other operation to be executed.

required
parameters Optional[Dict[str, Any]]

The parameters for the operation.

None
size Optional[int]

The number of results to return; if None or 0, uses the value of fetch_size configured on the block.

None
**execution_options Dict[str, Any]

Options to pass to Connection.execution_options.

{}

Returns:

Type Description
List[Tuple[Any]]

A list of tuples containing the data returned by the database, where each row is a tuple and each column is a value in the tuple.

Examples:

Create a table, insert three rows into it, and fetch two rows repeatedly.

from prefect_sqlalchemy import SqlAlchemyConnector

with SqlAlchemyConnector.load("MY_BLOCK") as database:
    database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
    database.execute_many(
        "INSERT INTO customers (name, address) VALUES (:name, :address);",
        seq_of_parameters=[
            {"name": "Ford", "address": "Highway 42"},
            {"name": "Unknown", "address": "Space"},
            {"name": "Me", "address": "Myway 88"},
        ],
    )
    results = database.fetch_many("SELECT * FROM customers", size=2)
    print(results)
    results = database.fetch_many("SELECT * FROM customers", size=2)
    print(results)

Source code in prefect_sqlalchemy/database.py
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
@sync_compatible
async def fetch_many(
    self,
    operation: str,
    parameters: Optional[Dict[str, Any]] = None,
    size: Optional[int] = None,
    **execution_options: Dict[str, Any],
) -> List[Tuple[Any]]:
    """
    Fetch a limited number of results from the database.

    Repeated calls using the same inputs to *any* of the fetch methods of this
    block will skip executing the operation again, and instead,
    return the next set of results from the previous execution,
    until the reset_cursors method is called.

    Args:
        operation: The SQL query or other operation to be executed.
        parameters: The parameters for the operation.
        size: The number of results to return; if None or 0, uses the value of
            `fetch_size` configured on the block.
        **execution_options: Options to pass to `Connection.execution_options`.

    Returns:
        A list of tuples containing the data returned by the database,
            where each row is a tuple and each column is a value in the tuple.

    Examples:
        Create a table, insert three rows into it, and fetch two rows repeatedly.
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        with SqlAlchemyConnector.load("MY_BLOCK") as database:
            database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
            database.execute_many(
                "INSERT INTO customers (name, address) VALUES (:name, :address);",
                seq_of_parameters=[
                    {"name": "Ford", "address": "Highway 42"},
                    {"name": "Unknown", "address": "Space"},
                    {"name": "Me", "address": "Myway 88"},
                ],
            )
            results = database.fetch_many("SELECT * FROM customers", size=2)
            print(results)
            results = database.fetch_many("SELECT * FROM customers", size=2)
            print(results)
        ```
    """  # noqa
    result_set = await self._get_result_set(
        text(operation), parameters, execution_options=execution_options
    )
    size = size or self.fetch_size
    self.logger.debug(f"Preparing to fetch {size} rows.")
    rows = result_set.fetchmany(size=size)
    return rows

fetch_one async

Fetch a single result from the database.

Repeated calls using the same inputs to any of the fetch methods of this block will skip executing the operation again, and instead, return the next set of results from the previous execution, until the reset_cursors method is called.

Parameters:

Name Type Description Default
operation str

The SQL query or other operation to be executed.

required
parameters Optional[Dict[str, Any]]

The parameters for the operation.

None
**execution_options Dict[str, Any]

Options to pass to Connection.execution_options.

{}

Returns:

Type Description
Tuple[Any]

A list of tuples containing the data returned by the database, where each row is a tuple and each column is a value in the tuple.

Examples:

Create a table, insert three rows into it, and fetch a row repeatedly.

from prefect_sqlalchemy import SqlAlchemyConnector

with SqlAlchemyConnector.load("MY_BLOCK") as database:
    database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
    database.execute_many(
        "INSERT INTO customers (name, address) VALUES (:name, :address);",
        seq_of_parameters=[
            {"name": "Ford", "address": "Highway 42"},
            {"name": "Unknown", "address": "Space"},
            {"name": "Me", "address": "Myway 88"},
        ],
    )
    results = True
    while results:
        results = database.fetch_one("SELECT * FROM customers")
        print(results)

Source code in prefect_sqlalchemy/database.py
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
@sync_compatible
async def fetch_one(
    self,
    operation: str,
    parameters: Optional[Dict[str, Any]] = None,
    **execution_options: Dict[str, Any],
) -> Tuple[Any]:
    """
    Fetch a single result from the database.

    Repeated calls using the same inputs to *any* of the fetch methods of this
    block will skip executing the operation again, and instead,
    return the next set of results from the previous execution,
    until the reset_cursors method is called.

    Args:
        operation: The SQL query or other operation to be executed.
        parameters: The parameters for the operation.
        **execution_options: Options to pass to `Connection.execution_options`.

    Returns:
        A list of tuples containing the data returned by the database,
            where each row is a tuple and each column is a value in the tuple.

    Examples:
        Create a table, insert three rows into it, and fetch a row repeatedly.
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        with SqlAlchemyConnector.load("MY_BLOCK") as database:
            database.execute("CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);")
            database.execute_many(
                "INSERT INTO customers (name, address) VALUES (:name, :address);",
                seq_of_parameters=[
                    {"name": "Ford", "address": "Highway 42"},
                    {"name": "Unknown", "address": "Space"},
                    {"name": "Me", "address": "Myway 88"},
                ],
            )
            results = True
            while results:
                results = database.fetch_one("SELECT * FROM customers")
                print(results)
        ```
    """  # noqa
    result_set = await self._get_result_set(
        text(operation), parameters, execution_options=execution_options
    )
    self.logger.debug("Preparing to fetch one row.")
    row = result_set.fetchone()
    return row

get_client

Returns either an engine or connection that can be used to query from databases.

Parameters:

Name Type Description Default
client_type Literal['engine', 'connection']

Select from either 'engine' or 'connection'.

required
**get_client_kwargs Dict[str, Any]

Additional keyword arguments to pass to either get_engine or get_connection.

{}

Returns:

Type Description
Union[Engine, AsyncEngine, Connection, AsyncConnection]

The authenticated SQLAlchemy engine or connection.

Examples:

Create an engine.

from prefect_sqlalchemy import SqlalchemyConnector

sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
engine = sqlalchemy_connector.get_client(client_type="engine")

Create a context managed connection.

from prefect_sqlalchemy import SqlalchemyConnector

sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
with sqlalchemy_connector.get_client(client_type="connection") as conn:
    ...

Source code in prefect_sqlalchemy/database.py
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
def get_client(
    self,
    client_type: Literal["engine", "connection"],
    **get_client_kwargs: Dict[str, Any],
) -> Union[Engine, AsyncEngine, Connection, AsyncConnection]:
    """
    Returns either an engine or connection that can be used to query from databases.

    Args:
        client_type: Select from either 'engine' or 'connection'.
        **get_client_kwargs: Additional keyword arguments to pass to
            either `get_engine` or `get_connection`.

    Returns:
        The authenticated SQLAlchemy engine or connection.

    Examples:
        Create an engine.
        ```python
        from prefect_sqlalchemy import SqlalchemyConnector

        sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
        engine = sqlalchemy_connector.get_client(client_type="engine")
        ```

        Create a context managed connection.
        ```python
        from prefect_sqlalchemy import SqlalchemyConnector

        sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
        with sqlalchemy_connector.get_client(client_type="connection") as conn:
            ...
        ```
    """  # noqa: E501
    if client_type == "engine":
        client = self.get_engine(**get_client_kwargs)
    elif client_type == "connection":
        client = self.get_connection(**get_client_kwargs)
    else:
        raise ValueError(
            f"{client_type!r} is not supported; choose from engine or connection."
        )
    return client

get_connection

Returns a connection that can be used to query from databases.

Parameters:

Name Type Description Default
begin bool

Whether to begin a transaction on the connection; if True, if any operations fail, the entire transaction will be rolled back.

True
**connect_kwargs Dict[str, Any]

Additional keyword arguments to pass to either engine.begin or engine.connect`.

{}

Returns:

Type Description
Union[Connection, AsyncConnection]

The SQLAlchemy Connection / AsyncConnection.

Examples:

Create an synchronous connection as a context-managed transaction.

from prefect_sqlalchemy import SqlAlchemyConnector

sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
with sqlalchemy_connector.get_connection(begin=False) as connection:
    connection.execute("SELECT * FROM table LIMIT 1;")

Create an asynchronous connection as a context-managed transacation.

import asyncio
from prefect_sqlalchemy import SqlAlchemyConnector

sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
async with sqlalchemy_connector.get_connection(begin=False) as connection:
    asyncio.run(connection.execute("SELECT * FROM table LIMIT 1;"))

Source code in prefect_sqlalchemy/database.py
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
def get_connection(
    self, begin: bool = True, **connect_kwargs: Dict[str, Any]
) -> Union[Connection, AsyncConnection]:
    """
    Returns a connection that can be used to query from databases.

    Args:
        begin: Whether to begin a transaction on the connection; if True, if
            any operations fail, the entire transaction will be rolled back.
        **connect_kwargs: Additional keyword arguments to pass to either
            `engine.begin` or engine.connect`.

    Returns:
        The SQLAlchemy Connection / AsyncConnection.

    Examples:
        Create an synchronous connection as a context-managed transaction.
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
        with sqlalchemy_connector.get_connection(begin=False) as connection:
            connection.execute("SELECT * FROM table LIMIT 1;")
        ```

        Create an asynchronous connection as a context-managed transacation.
        ```python
        import asyncio
        from prefect_sqlalchemy import SqlAlchemyConnector

        sqlalchemy_connector = SqlAlchemyConnector.load("BLOCK_NAME")
        async with sqlalchemy_connector.get_connection(begin=False) as connection:
            asyncio.run(connection.execute("SELECT * FROM table LIMIT 1;"))
        ```
    """  # noqa: E501
    engine = self.get_engine()
    if begin:
        connection = engine.begin(**connect_kwargs)
    else:
        connection = engine.connect(**connect_kwargs)
    self.logger.info("Created a new connection.")
    return connection

get_engine

Returns an authenticated engine that can be used to query from databases.

If an existing engine exists, return that one.

Returns:

Type Description
Union[Engine, AsyncEngine]

The authenticated SQLAlchemy Engine / AsyncEngine.

Examples:

Create an asynchronous engine to PostgreSQL using URL params.

from prefect import flow
from prefect_sqlalchemy import (
    SqlAlchemyConnector, ConnectionComponents, AsyncDriver
)

@flow
def sqlalchemy_credentials_flow():
    sqlalchemy_credentials = SqlAlchemyConnector(
    connection_info=ConnectionComponents(
            driver=AsyncDriver.POSTGRESQL_ASYNCPG,
            username="prefect",
            password="prefect_password",
            database="postgres"
        )
    )
    print(sqlalchemy_credentials.get_engine())

sqlalchemy_credentials_flow()

Create a synchronous engine to Snowflake using the url kwarg.

from prefect import flow
from prefect_sqlalchemy import SqlAlchemyConnector, AsyncDriver

@flow
def sqlalchemy_credentials_flow():
    url = (
        "snowflake://<user_login_name>:<password>"
        "@<account_identifier>/<database_name>"
        "?warehouse=<warehouse_name>"
    )
    sqlalchemy_credentials = SqlAlchemyConnector(url=url)
    print(sqlalchemy_credentials.get_engine())

sqlalchemy_credentials_flow()

Source code in prefect_sqlalchemy/database.py
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
def get_engine(
    self, **create_engine_kwargs: Dict[str, Any]
) -> Union[Engine, AsyncEngine]:
    """
    Returns an authenticated engine that can be
    used to query from databases.

    If an existing engine exists, return that one.

    Returns:
        The authenticated SQLAlchemy Engine / AsyncEngine.

    Examples:
        Create an asynchronous engine to PostgreSQL using URL params.
        ```python
        from prefect import flow
        from prefect_sqlalchemy import (
            SqlAlchemyConnector, ConnectionComponents, AsyncDriver
        )

        @flow
        def sqlalchemy_credentials_flow():
            sqlalchemy_credentials = SqlAlchemyConnector(
            connection_info=ConnectionComponents(
                    driver=AsyncDriver.POSTGRESQL_ASYNCPG,
                    username="prefect",
                    password="prefect_password",
                    database="postgres"
                )
            )
            print(sqlalchemy_credentials.get_engine())

        sqlalchemy_credentials_flow()
        ```

        Create a synchronous engine to Snowflake using the `url` kwarg.
        ```python
        from prefect import flow
        from prefect_sqlalchemy import SqlAlchemyConnector, AsyncDriver

        @flow
        def sqlalchemy_credentials_flow():
            url = (
                "snowflake://<user_login_name>:<password>"
                "@<account_identifier>/<database_name>"
                "?warehouse=<warehouse_name>"
            )
            sqlalchemy_credentials = SqlAlchemyConnector(url=url)
            print(sqlalchemy_credentials.get_engine())

        sqlalchemy_credentials_flow()
        ```
    """
    if self._engine is not None:
        self.logger.debug("Reusing existing engine.")
        return self._engine

    engine_kwargs = dict(
        url=self._rendered_url,
        connect_args=self.connect_args or {},
        **create_engine_kwargs,
    )
    if self._driver_is_async:
        # no need to await here
        engine = create_async_engine(**engine_kwargs)
    else:
        engine = create_engine(**engine_kwargs)
    self.logger.info("Created a new engine.")

    if self._engine is None:
        self._engine = engine

    return engine

reset_async_connections async

Tries to close all opened connections and their results.

Examples:

Resets connections so fetch_* methods return new results.

import asyncio
from prefect_sqlalchemy import SqlAlchemyConnector

async def example_run():
    async with SqlAlchemyConnector.load("MY_BLOCK") as database:
        results = await database.fetch_one("SELECT * FROM customers")
        await database.reset_async_connections()
        results = await database.fetch_one("SELECT * FROM customers")

asyncio.run(example_run())

Source code in prefect_sqlalchemy/database.py
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
async def reset_async_connections(self) -> None:
    """
    Tries to close all opened connections and their results.

    Examples:
        Resets connections so `fetch_*` methods return new results.
        ```python
        import asyncio
        from prefect_sqlalchemy import SqlAlchemyConnector

        async def example_run():
            async with SqlAlchemyConnector.load("MY_BLOCK") as database:
                results = await database.fetch_one("SELECT * FROM customers")
                await database.reset_async_connections()
                results = await database.fetch_one("SELECT * FROM customers")

        asyncio.run(example_run())
        ```
    """
    if not self._driver_is_async:
        raise RuntimeError(
            f"{self._rendered_url.drivername} has no asynchronous connections. "
            f"Please use the `reset_connections` method instead."
        )

    if self._exit_stack is None:
        self.logger.info("There were no connections to reset.")
        return

    self._reset_cursor_results()
    await self._exit_stack.aclose()
    self.logger.info("Reset opened connections and their results.")

reset_connections async

Tries to close all opened connections and their results.

Examples:

Resets connections so fetch_* methods return new results.

from prefect_sqlalchemy import SqlAlchemyConnector

with SqlAlchemyConnector.load("MY_BLOCK") as database:
    results = database.fetch_one("SELECT * FROM customers")
    database.reset_connections()
    results = database.fetch_one("SELECT * FROM customers")

Source code in prefect_sqlalchemy/database.py
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
@sync_compatible
async def reset_connections(self) -> None:
    """
    Tries to close all opened connections and their results.

    Examples:
        Resets connections so `fetch_*` methods return new results.
        ```python
        from prefect_sqlalchemy import SqlAlchemyConnector

        with SqlAlchemyConnector.load("MY_BLOCK") as database:
            results = database.fetch_one("SELECT * FROM customers")
            database.reset_connections()
            results = database.fetch_one("SELECT * FROM customers")
        ```
    """
    if self._driver_is_async:
        raise RuntimeError(
            f"{self._rendered_url.drivername} has no synchronous connections. "
            f"Please use the `reset_async_connections` method instead."
        )

    if self._exit_stack is None:
        self.logger.info("There were no connections to reset.")
        return

    self._reset_cursor_results()
    self._exit_stack.close()
    self.logger.info("Reset opened connections and their results.")

sqlalchemy_execute async

Executes a SQL DDL or DML statement; useful for creating tables and inserting rows since this task does not return any objects.

Parameters:

Name Type Description Default
statement str

The statement to execute against the database.

required
sqlalchemy_credentials DatabaseCredentials

The credentials to use to authenticate.

required
params Optional[Union[Tuple[Any], Dict[str, Any]]]

The params to replace the placeholders in the query.

None

Examples:

Create table named customers and insert values.

from prefect_sqlalchemy import DatabaseCredentials, AsyncDriver
from prefect_sqlalchemy.database import sqlalchemy_execute
from prefect import flow

@flow
def sqlalchemy_execute_flow():
    sqlalchemy_credentials = DatabaseCredentials(
        driver=AsyncDriver.SQLITE_AIOSQLITE,
        database="prefect.db",
    )
    sqlalchemy_execute(
        "CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);",
        sqlalchemy_credentials,
    )
    sqlalchemy_execute(
        "INSERT INTO customers (name, address) VALUES (:name, :address);",
        sqlalchemy_credentials,
        params={"name": "Marvin", "address": "Highway 42"}
    )

sqlalchemy_execute_flow()

Source code in prefect_sqlalchemy/database.py
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@task
async def sqlalchemy_execute(
    statement: str,
    sqlalchemy_credentials: "DatabaseCredentials",
    params: Optional[Union[Tuple[Any], Dict[str, Any]]] = None,
):
    """
    Executes a SQL DDL or DML statement; useful for creating tables and inserting rows
    since this task does not return any objects.

    Args:
        statement: The statement to execute against the database.
        sqlalchemy_credentials: The credentials to use to authenticate.
        params: The params to replace the placeholders in the query.

    Examples:
        Create table named customers and insert values.
        ```python
        from prefect_sqlalchemy import DatabaseCredentials, AsyncDriver
        from prefect_sqlalchemy.database import sqlalchemy_execute
        from prefect import flow

        @flow
        def sqlalchemy_execute_flow():
            sqlalchemy_credentials = DatabaseCredentials(
                driver=AsyncDriver.SQLITE_AIOSQLITE,
                database="prefect.db",
            )
            sqlalchemy_execute(
                "CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);",
                sqlalchemy_credentials,
            )
            sqlalchemy_execute(
                "INSERT INTO customers (name, address) VALUES (:name, :address);",
                sqlalchemy_credentials,
                params={"name": "Marvin", "address": "Highway 42"}
            )

        sqlalchemy_execute_flow()
        ```
    """
    warnings.warn(
        "sqlalchemy_query is now deprecated and will be removed March 2023; "
        "please use SqlAlchemyConnector execute_* methods instead.",
        DeprecationWarning,
    )
    # do not return anything or else results in the error:
    # This result object does not return rows. It has been closed automatically
    engine = sqlalchemy_credentials.get_engine()
    async_supported = sqlalchemy_credentials._driver_is_async
    async with _connect(engine, async_supported) as connection:
        await _execute(connection, statement, params, async_supported)

sqlalchemy_query async

Executes a SQL query; useful for querying data from existing tables.

Parameters:

Name Type Description Default
query str

The query to execute against the database.

required
sqlalchemy_credentials DatabaseCredentials

The credentials to use to authenticate.

required
params Optional[Union[Tuple[Any], Dict[str, Any]]]

The params to replace the placeholders in the query.

None
limit Optional[int]

The number of rows to fetch. Note, this parameter is executed on the client side, i.e. passed to fetchmany. To limit on the server side, add the LIMIT clause, or the dialect's equivalent clause, like TOP, to the query.

None

Returns:

Type Description
List[Tuple[Any]]

The fetched results.

Examples:

Query postgres table with the ID value parameterized.

from prefect_sqlalchemy import DatabaseCredentials, AsyncDriver
from prefect_sqlalchemy.database import sqlalchemy_query
from prefect import flow

@flow
def sqlalchemy_query_flow():
    sqlalchemy_credentials = DatabaseCredentials(
        driver=AsyncDriver.SQLITE_AIOSQLITE,
        database="prefect.db",
    )
    result = sqlalchemy_query(
        "SELECT * FROM customers WHERE name = :name;",
        sqlalchemy_credentials,
        params={"name": "Marvin"},
    )
    return result

sqlalchemy_query_flow()

Source code in prefect_sqlalchemy/database.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@task
async def sqlalchemy_query(
    query: str,
    sqlalchemy_credentials: "DatabaseCredentials",
    params: Optional[Union[Tuple[Any], Dict[str, Any]]] = None,
    limit: Optional[int] = None,
) -> List[Tuple[Any]]:
    """
    Executes a SQL query; useful for querying data from existing tables.

    Args:
        query: The query to execute against the database.
        sqlalchemy_credentials: The credentials to use to authenticate.
        params: The params to replace the placeholders in the query.
        limit: The number of rows to fetch. Note, this parameter is
            executed on the client side, i.e. passed to `fetchmany`.
            To limit on the server side, add the `LIMIT` clause, or
            the dialect's equivalent clause, like `TOP`, to the query.

    Returns:
        The fetched results.

    Examples:
        Query postgres table with the ID value parameterized.
        ```python
        from prefect_sqlalchemy import DatabaseCredentials, AsyncDriver
        from prefect_sqlalchemy.database import sqlalchemy_query
        from prefect import flow

        @flow
        def sqlalchemy_query_flow():
            sqlalchemy_credentials = DatabaseCredentials(
                driver=AsyncDriver.SQLITE_AIOSQLITE,
                database="prefect.db",
            )
            result = sqlalchemy_query(
                "SELECT * FROM customers WHERE name = :name;",
                sqlalchemy_credentials,
                params={"name": "Marvin"},
            )
            return result

        sqlalchemy_query_flow()
        ```
    """
    warnings.warn(
        "sqlalchemy_query is now deprecated and will be removed March 2023; "
        "please use SqlAlchemyConnector fetch_* methods instead.",
        DeprecationWarning,
    )
    engine = sqlalchemy_credentials.get_engine()
    async_supported = sqlalchemy_credentials._driver_is_async
    async with _connect(engine, async_supported) as connection:
        result = await _execute(connection, query, params, async_supported)
        # some databases, like sqlite, require a connection still open to fetch!
        rows = result.fetchall() if limit is None else result.fetchmany(limit)
    return rows