from prefect import flow
from prefect_airbyte.connections import trigger_sync
from prefect_dbt.cli.commands import trigger_dbt_cli_command

@flow(name='Airbyte')
def run_airbyte_syncs():
    airbyte_sync = trigger_sync (
        airbyte_server_host='localhost',
        airbyte_server_port=8000,
        connection_id='5fdd8809-b7c6-4568-a6d2-72e17b648bb1',
        poll_interval_s=3,
        timeout=1800,
        status_updates=True
    )

    return airbyte_sync

@flow(name='dbt')
def run_dbt_commands():
    dbt_result = trigger_dbt_cli_command(command='dbt run',
        profiles_dir='C:\\Users\\JSchueller\\.dbt',
        project_dir='C:\\Users\\JSchueller\\etl-architecture-demo\\etl_architecture_demo',
        shell='powershell',
        return_all=True
    )
    
    return dbt_result

@flow(name='execute')
def execute():
    # run_airbyte_syncs()
    run_dbt_commands()

if __name__ == '__main__':
    execute()

10:37:02.302 | INFO    | prefect.engine - Created flow run 'rose-zebu' for flow 'execute'
10:37:03.752 | INFO    | Flow run 'rose-zebu' - Created subflow run 'icy-wallaby' for flow 'dbt'
10:37:03.936 | INFO    | Flow run 'icy-wallaby' - Created task run 'trigger_dbt_cli_command-321ca940-0' for task 'trigger_dbt_cli_command'
10:37:03.937 | INFO    | Flow run 'icy-wallaby' - Executing 'trigger_dbt_cli_command-321ca940-0' immediately...
10:37:04.036 | INFO    | Task run 'trigger_dbt_cli_command-321ca940-0' - Running dbt command: dbt run --profiles-dir C:\Users\JSchueller\.dbt --project-dir C:\Users\JSchueller\etl-architecture-demo\etl_architecture_demo
10:37:09.409 | INFO    | Task run 'trigger_dbt_cli_command-321ca940-0' - Finished in state Completed()
10:37:09.508 | INFO    | Flow run 'icy-wallaby' - Finished in state Completed()
10:37:09.581 | INFO    | Flow run 'rose-zebu' - Finished in state Completed('All states completed.')

  ~   prefect-env 3.9.12  dbt run --profiles-dir 'C:\Users\JSchueller\.dbt' --project-dir 'C:\Users\JSchueller\etl-architecture-demo\etl_architecture_demo'
14:25:04  Running with dbt=1.2.1
14:25:04  Unable to do partial parsing because of a dbt version mismatch. Saved manifest version: 1.1.2. Current version: 1.2.1.
14:25:05  Found 3 models, 11 tests, 0 snapshots, 0 analyses, 267 macros, 0 operations, 0 seed files, 2 sources, 0 exposures, 0 metrics
14:25:05
14:25:09  Concurrency: 1 threads (target='dev')
14:25:09
14:25:09  1 of 3 START view model dbt.stg_accounts ....................................... [RUN]
14:25:11  1 of 3 OK created view model dbt.stg_accounts .................................. [SUCCESS 1 in 1.29s]
14:25:11  2 of 3 START view model dbt.stg_contacts ....................................... [RUN]
14:25:12  2 of 3 OK created view model dbt.stg_contacts .................................. [SUCCESS 1 in 1.27s]
14:25:12  3 of 3 START table model dbt.dim_accounts ...................................... [RUN]
14:25:23  3 of 3 OK created table model dbt.dim_accounts ................................. [SUCCESS 1 in 11.48s]
14:25:23
14:25:23  Finished running 2 view models, 1 table model in 0 hours 0 minutes and 17.79 seconds (17.79s).
14:25:23
14:25:23  Completed successfully
14:25:23
14:25:23  Done. PASS=3 WARN=0 ERROR=0 SKIP=0 TOTAL=3

shell_run_command

import asyncio
from prefect_shell.commands import *

c = "echo 'Hello world!'"

asyncio.run(shell_run_command.fn(command=c, shell='powershell', return_all=True))

powershell prefect-temporary.sh

from anyio import open_process

stream_level=logging.DEBUG

with tempfile.NamedTemporaryFile(prefix="prefect-") as tmp:
        if helper_command:
            tmp.write(helper_command.encode())
            tmp.write(os.linesep.encode())
        tmp.write(command.encode())
        tmp.flush()

if sys.platform == "win32":
            shell_command = " ".join(shell_command)