subprocess.check_call("prefect --no-prompt deploy --all".split())

import asyncio
import os
import pathlib
import subprocess
import yaml
from typing import Union

from prefect.cli.deploy import deploy


class FlowDeployer:
    DEPLOYMENT_FILENAME = 'prefect.yaml'
    #DEPLOYMENT_BATCH = "prefect_deploy.bat"

    def __init__(self, parent_folder_path: Union[str, pathlib.Path],
                 deployment_name: str = None, root_folder: Union[str, pathlib.Path] = None):
        self.deployment_name = deployment_name
        self.root_folder = pathlib.Path(__file__).resolve().parent if not root_folder else root_folder
        self.parent_folder_path = self.root_folder / parent_folder_path

    @property
    def parent_folder_path(self):
        return self._parent_folder_path

    @parent_folder_path.setter
    def parent_folder_path(self, v):
        if isinstance(v, str):
            v = pathlib.Path(v)
        self._parent_folder_path = v

    async def deploy_flows_from_parent_folder(self):
        try:
            path = self.parent_folder_path
            if isinstance(path, str):
                path = pathlib.Path(path)
            parent_deployment_file = path.joinpath(self.DEPLOYMENT_FILENAME)
            if not parent_deployment_file.is_file():
                raise Exception(f"In questo folder {path} non c'è il file {self.DEPLOYMENT_FILENAME}")
            pj_root_deployment_file = self.root_folder.joinpath(self.DEPLOYMENT_FILENAME)
            if pj_root_deployment_file.is_file():
                os.remove(pj_root_deployment_file)
            with open(parent_deployment_file) as f:
                yaml_file = yaml.safe_load(f)
            for d in yaml_file['deployments']:
                d['entrypoint'] = str(path.joinpath(d['entrypoint'])).replace('\\', '/')
            with open(pj_root_deployment_file, 'w') as f:
                yaml.dump(yaml_file, f, sort_keys=False)
            await deploy(deploy_all=True)
            #here the old way to deploy all that we want to remove
            #cmd = 'call "venv/scripts/activate" \n'
            #cmd += f"prefect deploy --all"

            #with open(self.DEPLOYMENT_BATCH, 'w') as f:
                #f.write(cmd)
            #subprocess.check_output(self.DEPLOYMENT_BATCH)
        except Exception as ex:
            raise ex

    def execute(self):
        asyncio.run(self.deploy_flows_from_parent_folder())


if __name__ == "__main__":
    parent_folder = pathlib.Path('./customers/****')
    root_folder = pathlib.Path('C:/prefect')
    d = FlowDeployer(parent_folder_path=parent_folder, root_folder=root_folder)
    d.execute()