def__exit__(self,exc_type,exc_val,exc_tb)->None:# type: ignoreifexc_typeisnotNone:error_types={Stage.deploy:DeployError,Stage.execute:ExecuteError}raiseerror_types[self._mode](f"Oops... failed during: {self._mode}")fromexc_valiflen(self._project.workflows)==0:_ilog.info("Doing nothing no workflows...")returnifself._mode==Stage.unittest:# Mode is purely for testing purposes for the _Project internal class_ilog.info("Running unit tests...")returnifself._mode.value==Stage.deploy.value:_ilog.info("Deploying changes... to %s",ctx.env)ifself.codegen_mechanismisNone:raiseValueError("codegen_mechanism cannot be None; please raise a github issue for this.")codegen=self.codegen_mechanism(project=self._project,id_=f"{ctx.env}_{self.name}",env=ctx.env,**(self.codegen_kwargsor{}),)codegen.synth()ifself._mode.value==Stage.execute.value:wf_id=ctx.get_parameter(BrickflowInternalVariables.workflow_id.value,self.debug_execute_workflow,)t_id=ctx.get_parameter(BrickflowInternalVariables.task_id.value,self.debug_execute_task)ifwf_idisNoneort_idisNone:_ilog.info("No workflow id or task key was able to found; doing nothing...")returnworkflow=self._project.get_workflow(wf_id)task=workflow.get_task(t_id)task.execute()
def__post_init__(self)->None:# during entry of project enable logging_ilog.setLevel(logging.INFO)self._mode=Stage[config(BrickflowEnvVars.BRICKFLOW_MODE.value,default=Stage.execute.value)]self.entry_point_path=self.entry_point_pathorget_caller_info()deploy_settings=BrickflowProjectDeploymentSettings()# setup current_projectenv_project_name=deploy_settings.brickflow_project_nameself.libraries=self.librariesor[]ifdeploy_settings.brickflow_auto_add_librariesisTrue:_ilog.info("Auto adding brickflow libraries...")self.libraries=filter_bf_related_libraries(self.libraries)# here libraries should not be null anymore if this branch is invokedself.libraries+=get_brickflow_libraries()if(env_project_nameisnotNoneandself.nameisnotNoneandenv_project_name!=self.name):raiseValueError("Project name in config files and entrypoint must be the same")ctx.set_current_project(self.nameorenv_project_name)# always setup first# populate bundle info via env varsself.bundle_obj_name=config(BrickflowEnvVars.BRICKFLOW_BUNDLE_OBJ_NAME.value,default=".brickflow_bundles",)self.bundle_base_path=config(BrickflowEnvVars.BRICKFLOW_BUNDLE_BASE_PATH.value,default="/Users/${workspace.current_user.userName}",)self.git_reference=config(BrickflowEnvVars.BRICKFLOW_GIT_REF.value,default=self.get_git_ref())if(self._mode==Stage.deployandctx.is_local()isFalseandself.git_referenceisNone):raiseValueError("git_reference must be set when deploying to non-local envs")self.provider=config(BrickflowEnvVars.BRICKFLOW_GIT_PROVIDER.value,default=self.provider)self.git_repo=config(BrickflowEnvVars.BRICKFLOW_GIT_REPO.value,default=self.git_repo)deployment_mode=config(BrickflowEnvVars.BRICKFLOW_DEPLOYMENT_MODE.value,default="bundle")ifdeployment_mode==BrickflowDeployMode.BUNDLE.value:self.codegen_mechanism=DatabricksBundleCodegenifself.codegen_kwargsisNone:self.codegen_kwargs={}
defget_git_ref(self)->Optional[str]:ifself._mode==Stage.deploy:ifself.git_referenceisnotNone:returnself.git_referenceelse:try:returnf"commit/{get_current_commit()}"exceptException:_ilog.warning("Unable to get current commit; defaulting to empty string")return"commit/fake-local-stub"ifctx.is_local()elseNoneelse:returnself.git_referenceifself.git_referenceisnotNoneelse""