Repository: PrefectHQ/prefect
Issue category: Feature Request

Issue:
return `CursorResult` from prefect_sqlalchemy execute/execute_many methods
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.

### Prefect Version

3.x

### Describe the current behavior

`execute_many` method returns `None`

### Describe the proposed behavior

`execute_many` method would return `CursorResult` so that users could interact with the result of the executed query

### Example Use

_No response_

### Additional context

_No response_


Hints:

Patch:
diff --git a/src/integrations/prefect-sqlalchemy/prefect_sqlalchemy/database.py b/src/integrations/prefect-sqlalchemy/prefect_sqlalchemy/database.py
index bdda680e3772..f98b109fb371 100644
--- a/src/integrations/prefect-sqlalchemy/prefect_sqlalchemy/database.py
+++ b/src/integrations/prefect-sqlalchemy/prefect_sqlalchemy/database.py
@@ -609,7 +609,7 @@ async def execute(
         operation: str,
         parameters: Optional[Dict[str, Any]] = None,
         **execution_options: Dict[str, Any],
-    ) -> None:
+    ) -> CursorResult:
         """
         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.
@@ -636,13 +636,14 @@ async def execute(
             ```
         """  # noqa
         async with self._manage_connection(begin=False) as connection:
-            await self._async_sync_execute(
+            result = await self._async_sync_execute(
                 connection,
                 text(operation),
                 parameters,
                 execution_options=execution_options,
             )
         self.logger.info(f"Executed the operation, {operation!r}")
+        return result
 
     @sync_compatible
     async def execute_many(
@@ -650,7 +651,7 @@ async def execute_many(
         operation: str,
         seq_of_parameters: List[Dict[str, Any]],
         **execution_options: Dict[str, Any],
-    ) -> None:
+    ) -> CursorResult:
         """
         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.
@@ -681,7 +682,7 @@ async def execute_many(
             ```
         """  # noqa
         async with self._manage_connection(begin=False) as connection:
-            await self._async_sync_execute(
+            result = await self._async_sync_execute(
                 connection,
                 text(operation),
                 seq_of_parameters,
@@ -690,6 +691,7 @@ async def execute_many(
         self.logger.info(
             f"Executed {len(seq_of_parameters)} operations based off {operation!r}."
         )
+        return result
 
     async def __aenter__(self):
         """


Test patch:
diff --git a/src/integrations/prefect-sqlalchemy/tests/test_database.py b/src/integrations/prefect-sqlalchemy/tests/test_database.py
index bd17537ea2c2..a2f0327ace48 100644
--- a/src/integrations/prefect-sqlalchemy/tests/test_database.py
+++ b/src/integrations/prefect-sqlalchemy/tests/test_database.py
@@ -12,6 +12,7 @@
 )
 from sqlalchemy import __version__ as SQLALCHEMY_VERSION
 from sqlalchemy.engine import Connection, Engine
+from sqlalchemy.engine.cursor import CursorResult
 from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine
 
 from prefect import flow, task
@@ -156,14 +157,14 @@ async def connector_with_data(self, tmp_path, request):
             ),
             fetch_size=2,
         )
-        await credentials.execute(
+        create_result = await credentials.execute(
             "CREATE TABLE IF NOT EXISTS customers (name varchar, address varchar);"
         )
-        await credentials.execute(
+        insert_result = await credentials.execute(
             "INSERT INTO customers (name, address) VALUES (:name, :address);",
             parameters={"name": "Marvin", "address": "Highway 42"},
         )
-        await credentials.execute_many(
+        many_result = await credentials.execute_many(
             "INSERT INTO customers (name, address) VALUES (:name, :address);",
             seq_of_parameters=[
                 {"name": "Ford", "address": "Highway 42"},
@@ -171,6 +172,9 @@ async def connector_with_data(self, tmp_path, request):
                 {"name": "Me", "address": "Myway 88"},
             ],
         )
+        assert isinstance(many_result, CursorResult)
+        assert isinstance(insert_result, CursorResult)
+        assert isinstance(create_result, CursorResult)
         yield credentials
 
     @pytest.fixture(params=[True, False])