Docker Prefect

working_register.py

import os
from prefect import Flow
#from prefect.environments.storage import Docker
from prefect.tasks.shell import ShellTask
from prefect.tasks.secrets.base import PrefectSecret


import boto3
import docker
import base64

# Flow
from moc_data_flow import tsx_imb_fl


# prefect
project_name =  "market-on-close"

# Docker
dckr_image_name = "pretest"
dckr_tag = "latest"

working_dir_path = os.getcwd()
print(f"Working Dir: {working_dir_path}")
docker_flpth = os.path.join(working_dir_path, "Dockerfile")
print(f"Docker flpth: {docker_flpth}")

# aws
aws_ecr_repo_name = dckr_image_name
aws_region = "us-east-2"



############## Storage ecr docker flow ##############
# 1. Reset Auth (hackish)
dkr_ecr_scrt = PrefectSecret("docker_ecr_login").run()

get_ecr_auth_token = ShellTask(helper_script="cd ~")
ecr_auth_token = get_ecr_auth_token.run(command=dkr_ecr_scrt)

# Get ecr creds
ecr_client = boto3.client('ecr', region_name=aws_region)
ecr_token = ecr_client.get_authorization_token()

# # Decode the aws token
username, password = base64.b64decode(ecr_token['authorizationData'][0]['authorizationToken']).decode().split(':')
ecr_url = ecr_token['authorizationData'][0]['proxyEndpoint']

############################################################

# # # Registry URL for prefect or docker push
ecr_repo_name = f"{ecr_url.replace('https://', '')}"#/{aws_ecr_repo_name}" #:latest"


# 2. Add Docker push to docker repo
tsx_imb_fl.storage = Docker(
    registry_url=ecr_repo_name,
    python_dependencies=[
        "pandas", "sqlalchemy", "psycopg2",
        "beautifulsoup4","lxml", "boto3","requests", 
        "dask_cloudprovider"],
    dockerfile=docker_flpth,
    image_name=dckr_image_name,
    image_tag=dckr_tag,
    local_image=True
    )

# 3. Register 
pushlog = tsx_imb_fl.register(project_name="market-on-close", build=True)

print(pushlog)