from truefoundry.workflow import (
task,
workflow,
PythonTaskConfig,
TaskPythonBuild,
)
from truefoundry.deploy import Resources, NvidiaGPU
import requests
import json
cpu_config = PythonTaskConfig(
image=TaskPythonBuild(
python_version="3.9",
pip_packages=["truefoundry[workflow]==0.4.8"],
# requirements_path="requirements.txt"
),
resources=Resources(cpu_request=0.5)
)
gpu_config = PythonTaskConfig(
image=TaskPythonBuild(
python_version="3.9",
pip_packages=["truefoundry[workflow]==0.4.8", "pynvml==11.5.0"],
# requirements_path="requirements.txt",
cuda_version="11.5-cudnn8",
),
env={
"NVIDIA_DRIVER_CAPABILITIES": "compute,utility",
"NVIDIA_VISIBLE_DEVICES": "all",
},
resources=Resources(cpu_request=1.5, devices=[NvidiaGPU(name="T4", count=1)])
)
# Python Task: Fetch real-time prices for multiple cryptocurrencies from the CoinGecko API
@task(task_config=cpu_config)
def fetch_crypto_data() -> str:
response = requests.get(
"https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum,litecoin&vs_currencies=usd"
)
return response.text
# Python Task: Process the data to extract prices
@task(task_config=gpu_config)
def extract_prices(data: str) -> dict:
from pynvml import nvmlDeviceGetCount, nvmlInit
nvmlInit()
assert nvmlDeviceGetCount() > 0
json_data = json.loads(data)
prices = {
"bitcoin": json_data["bitcoin"]["usd"],
"ethereum": json_data["ethereum"]["usd"],
"litecoin": json_data["litecoin"]["usd"],
}
return prices
# Workflow: Combine all tasks
@workflow
def crypto_workflow() -> dict:
data = fetch_crypto_data()
prices = extract_prices(data=data)
return prices