import httpx
from datetime import timedelta
from prefect import flow, task
from prefect.tasks import task_input_hash


@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def get_url(url: str, params: dict = None):
    response = httpx.get(url, params=params)
    response.raise_for_status()
    return response.json()


def get_open_issues(repo_name: str, open_issues_count: int, per_page: int = 100):
    issues = []
    pages = range(1, -(open_issues_count // -per_page) + 1)
    for page in pages:
        issues.append(
            get_url.submit(
                f"<https://api.github.com/repos/{repo_name}/issues>",
                params={"page": page, "per_page": per_page, "state": "open"},
            )
        )
    return [i for p in issues for i in p.result()]


@flow(retries=3, retry_delay_seconds=5, log_prints=True)
def get_repo_info(repo_name: str = "PrefectHQ/prefect"):
    url = f"<https://api.github.com/repos/{repo_name}>"
    repo_stats = get_url(url)
    issues = get_open_issues(repo_name, repo_stats["open_issues_count"])
    issues_per_user = len(issues) / len(set([i["user"]["id"] for i in issues]))
    print(f"{repo_name} repository statistics 🤓:")
    print(f"Stars 🌠 : {repo_stats['stargazers_count']}")
    print(f"Forks 🍴 : {repo_stats['forks_count']}")
    print(f"Average open issues per user 💌 : {issues_per_user:.2f}")


if __name__ == "__main__":
    print("This is my-first-deployment")
    get_repo_info.serve(name="my-first-deployment")

prefect.yaml

-> cat prefect.yaml
# Welcome to your prefect.yaml file! You can use this file for storing and managing
# configuration for deploying your flows. We recommend committing this file to source
# control along with your flow code.

# Generic metadata about this project
name: workers_test
prefect-version: 2.14.9

build:

push:
- prefect_azure.deployments.steps.push_to_azure_blob_storage:
    id: push_code
    requires: prefect-azure>=0.2.8
    container: prefect-server
    folder: Deployments
    credentials: '{{ prefect.blocks.azure-blob-storage-credentials.prefect-server
      }}'

pull:
- prefect_azure.deployments.steps.pull_from_azure_blob_storage:
    id: pull_code
    requires: prefect-azure>=0.2.8
    container: '{{ push_code.container }}'
    folder: '{{ push_code.folder }}'
    credentials: '{{ prefect.blocks.azure-blob-storage-credentials.prefect-server
      }}'

deployments:
- name: prefect.yaml
  version:
  tags: []
  description:
  entrypoint: repo_info_v1.py:get_repo_info
  parameters: {}
  work_pool:
    name: arte-k8s
    work_queue_name: default
    job_variables:
      image: prefecthq/prefect:2.14.9-python3.11-kubernetes
        #image: "<http://docker-arte-release.dr-uw2.adobeitc.com/arte-flows-server:latest|docker-arte-release.dr-uw2.adobeitc.com/arte-flows-server:latest>"
        #image_pull_secrets: "artedocker"
  schedule:
  pull:
  - prefect_azure.deployments.steps.pull_from_azure_blob_storage:
      id: pull_code
      requires: prefect-azure>=0.2.8
      container: prefect-server
      folder: Deployments
      credentials: '{{ prefect.blocks.azure-blob-storage-credentials.prefect-server
        }}'