register blocks using seperate python files

make/04_setup_prefect.mk

@@ -1,9 +1,17 @@
 ##@ [Prefect: Setup]
 
+.PHONY: set-prefect-env-variarables
+set-prefect-env-variarables: ## Set Environment variables for Prefect
+	prefect config set PREFECT_LOGGING_EXTRA_LOGGERS=${PREFECT_LOGGING_EXTRA_LOGGERS}
+
 .PHONY: create-prefect-blocks
 create-prefect-blocks: ## Create Prefect Blocks
 	make env-init
-	python3 src/scripts/create_blocks.py
+	prefect block register --file src/blocks/create_gcp_credentials.py
+	prefect block register --file src/blocks/create_github.py
+	prefect block register --file src/blocks/create_google_cloud_run.py
+	prefect block register --file src/blocks/create_unsplash_access_key.py
+	prefect block register --file src/blocks/create_bigquery.py
 
 .PHONY: create-prefect-artifact-repository
 create-prefect-artifact-repository: ## Create GCP Artificat Repository for Prefect Flows
@@ -22,6 +30,8 @@ create-google-cloud-run-push-work-pool: ## Create Google Cloud Run Workflow (for
 
 .PHONY: setup-prefect
 setup-prefect: ## Setup Prefect Blocks, Artifact Repo and Work Pools	
+	@echo "Setting Environment variables for Prefect"
+	"$(MAKE)" set-prefect-env-variarables
 	@echo "Creating Prefect Blocks"
 	"$(MAKE)" create-prefect-blocks
 	@echo "Create Artificat Repository for Prefect Flows"

src/blocks/create_bigquery.py

@@ -0,0 +1,15 @@
+"""Programmatically create GCP Bigquery Block for Prefect"""
+
+from prefect_gcp import BigQueryWarehouse, GcpCredentials
+
+from src.utils import load_env_variables
+
+env_variables = load_env_variables()
+
+ENV = env_variables["ENV"]  # dev, test or prod
+PREFECT_BLOCK_NAME_GCP_CREDENTIALS = f"{env_variables['GCP_PROJECT_ID']}-{env_variables['GCP_DEPLOYMENT_SERVICE_ACCOUNT']}"
+GOOGLE_BIGQUERY_BLOCK_NAME = f"{env_variables['GCP_PROJECT_ID']}-bigquery-{ENV}"
+
+bigquery_block = BigQueryWarehouse(
+    gcp_credentials=GcpCredentials.load(PREFECT_BLOCK_NAME_GCP_CREDENTIALS)
+).save(name=GOOGLE_BIGQUERY_BLOCK_NAME, overwrite=True)

src/blocks/create_gcp_credentials.py

@@ -0,0 +1,16 @@
+"""Programmatically create GCP Credential Block for Prefect"""
+
+from prefect_gcp import GcpCredentials
+
+from src.utils import load_env_variables
+
+env_variables = load_env_variables()
+
+PREFECT_BLOCK_NAME_GCP_CREDENTIALS_BLOCK_NAME = f"{env_variables['GCP_PROJECT_ID']}-{env_variables['GCP_DEPLOYMENT_SERVICE_ACCOUNT']}"
+
+with open(f".secrets/deployment_sa_account.json", "r") as f:
+    service_account = f.read()
+
+gcp_credentials_block = GcpCredentials(service_account_info=service_account).save(
+    name=PREFECT_BLOCK_NAME_GCP_CREDENTIALS_BLOCK_NAME, overwrite=True
+)

src/blocks/create_github.py

@@ -0,0 +1,14 @@
+"""Programmatically create Github Block for Prefect"""
+
+from prefect.filesystems import GitHub
+from src.utils import load_env_variables
+
+env_vars = load_env_variables()
+ENV = env_vars["ENV"]  # dev, test or prod
+
+branches_mapping = {"dev": "develop", "test": "test", "prod": "master"}
+GITHUB_BLOCK_NAME = f"{env_vars['GCP_PROJECT_ID']}-github-{ENV}"
+
+github_block = GitHub(
+    repository=f"{env_vars['GITHUB_REPO']}", reference=branches_mapping[ENV]
+).save(name=GITHUB_BLOCK_NAME, overwrite=True)

src/blocks/create_google_cloud_run.py

@@ -0,0 +1,26 @@
+"""Programmatically create Google Cloud Run Block for Prefect"""
+
+from prefect_gcp import CloudRunJob, GcpCredentials
+
+from src.utils import load_env_variables
+
+env_vars = load_env_variables()
+ENV = env_vars["ENV"]  # dev, test or prod
+
+GCP_CREDENTIALS_PREFECT_BLOCK_NAME = (
+    f"{env_vars['GCP_PROJECT_ID']}-{env_vars['GCP_DEPLOYMENT_SERVICE_ACCOUNT']}"
+)
+GOOGLE_CLOUD_RUN_BLOCK_NAME = f"{env_vars['GCP_PROJECT_ID']}-google-cloud-run-{ENV}"
+
+gcp_credentials = GcpCredentials.load(GCP_CREDENTIALS_PREFECT_BLOCK_NAME)
+project_id = gcp_credentials.project
+registry_adress = (
+    f"{env_vars['GCP_DEFAULT_REGION']}-docker.pkg.dev/{project_id}/prefect-{ENV}"
+)
+
+cloud_run_job_block = CloudRunJob(
+    credentials=gcp_credentials,
+    project_id=project_id,
+    image=f"{registry_adress}/prefect:{env_vars['PREFECT_VERSION']}-python{env_vars['PYTHON_VERSION']}",
+    region=env_vars["GCP_DEFAULT_REGION"],
+).save(name=GOOGLE_CLOUD_RUN_BLOCK_NAME, overwrite=True)

src/blocks/create_unsplash_access_key.py

@@ -0,0 +1,12 @@
+"""Programmatically create Unsplash Access Key Secret Block for Prefect"""
+
+from prefect.blocks.system import Secret
+from src.utils import load_env_variables
+
+env_vars = load_env_variables()
+
+UNSPLASH_ACCESS_KEY_BLOCK_NAME = f"{env_vars['GCP_PROJECT_ID']}-unsplash-access-key"
+
+unsplash_access_key_block = Secret(value=env_vars["UNSPLASH_ACCESS_KEY"]).save(
+    name=UNSPLASH_ACCESS_KEY_BLOCK_NAME, overwrite=True
+)