Hi @davidgs !
Thank you so much for replying, sure…let me give you more information about the code.
About the python client I’m using, here is the github link to it:
However, I think the code lacks the part where we configure the connection protocol or the part where we place our input on the certificates that we are currently using. Here’s my code for the worker:
import logging
import os
from camunda.external_task.external_task import ExternalTask, TaskResult
from camunda.external_task.external_task_worker import ExternalTaskWorker
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from urllib.parse import quote, unquote_plus
from utilities.helpers import create_logger
from config import *
from models import *
logger_name = “wk-taskList”
logger = logging.getLogger(logger_name)
def handle_task(task: ExternalTask) → TaskResult:
“”"
This task handler you need to implement with your business logic.
After completion of business logic call either
task.complete() or task.failure() or task.bpmn_error()
to report status of task to Camunda
“”"
logger.debug(task)
conn_str_aml = (
f"postgresql+psycopg2://{SQL[‘AML’][‘USER’]}:%s"
f"@{SQL[‘AML’][‘HOSTNAME’]}:{SQL[‘AML’][‘PORT’]}"
f"/{SQL[‘AML’][‘DBNAME’]}"
)
conn_str_cmd = (
f"postgresql+psycopg2://{SQL['CMD']['USER']}:%s"
f"@{SQL['CMD']['HOSTNAME']}:{SQL['CMD']['PORT']}"
f"/{SQL['CMD']['DBNAME']}"
)
try:
engine_cmd = create_engine(conn_str_cmd % quote(SQL["CMD"]["PASS"]), echo=True)
connection_cmd = engine_cmd.connect()
Session_cmd = sessionmaker(engine_cmd)
session_cmd = Session_cmd()
// logic here
session_cmd.close()
connection_cmd.close()
engine_aml = create_engine(conn_str_aml % quote(SQL["AML"]["PASS"]), echo=True)
connection_aml = engine_aml.connect()
Session_aml = sessionmaker(engine_aml)
session_aml = Session_aml()
// logic here
session_aml.close()
connection_aml.close()
return task.complete({"status": "success"})
if name == “main”:
create_logger(logger_name, “wk-taskList”)
logger.info(f"Camunda Worker with ID: {WORKER_ID} started")
ExternalTaskWorker(worker_id=WORKER_ID, base_url=ENGINE_LOCAL_BASE_URL, config=WORKER_CONFIG).subscribe(
WORKER_TOPIC, handle_task
)
on the last portion, I did call the class ExternalTaskWorker and placed the params for the constructor.
ExternalTaskWorker(worker_id=WORKER_ID, base_url=ENGINE_LOCAL_BASE_URL, config=WORKER_CONFIG).subscribe(
WORKER_TOPIC, handle_task
)
I tried changing the value of the ENGINE_LOCAL_BASE_URL variable as well to reflect my camunda https URL but it did not work, so that clarifies that changing the URL will not solve the problem but rather there needs to be more that I’m not familiar with, as you mentioned maybe specifying the protocol or there needs to be an area where we define the certs.
Thank you so much for responding and in advance for the help!
