34 lines
908 B
Python
34 lines
908 B
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from app.jobs.base import BaseJob
|
|
from extensions.example.client import ExampleClient
|
|
|
|
|
|
logger = logging.getLogger("connecthub.extensions.example")
|
|
|
|
|
|
class ExampleJob(BaseJob):
|
|
job_id = "example.hello"
|
|
|
|
def run(self, params: dict[str, Any], secrets: dict[str, Any]) -> dict[str, Any]:
|
|
# params: 明文配置，例如 {"name": "Mars"}
|
|
name = params.get("name", "World")
|
|
|
|
# secrets: 解密后的明文，例如 {"token": "..."}
|
|
token = secrets.get("token", "<missing>")
|
|
|
|
client = ExampleClient(base_url="https://baidu.com", headers={"Authorization": f"Bearer {token}"})
|
|
try:
|
|
pong = client.ping()
|
|
|
|
finally:
|
|
client.close()
|
|
|
|
logger.info("ExampleJob ran name=%s pong=%s", name, pong)
|
|
return {"hello": name, "pong": pong}
|
|
|
|
|