From 4f6f5a9bbf60c7133080006beed06b2b228d1f1c Mon Sep 17 00:00:00 2001 From: yogeshwaran-c Date: Tue, 21 Apr 2026 16:46:56 +0530 Subject: [PATCH 1/3] feat(python): add return type hints to FlowProducer methods --- python/bullmq/flow_producer.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/python/bullmq/flow_producer.py b/python/bullmq/flow_producer.py index 928e3f26a4f..d56412f6348 100644 --- a/python/bullmq/flow_producer.py +++ b/python/bullmq/flow_producer.py @@ -1,4 +1,4 @@ -from typing import Union +from typing import Any, Optional, Union from bullmq.redis_connection import RedisConnection from bullmq.types import QueueBaseOptions from bullmq.scripts import Scripts @@ -43,17 +43,17 @@ def __init__(self, redisOpts: Union[dict, str] = {}, opts: QueueBaseOptions = {} self.scripts = Scripts( self.prefix, "__default__", self.redisConnection) - def queueFromNode(self, node:dict, queue_keys, prefix: str): + def queueFromNode(self, node: dict, queue_keys: QueueKeys, prefix: str) -> MinimalQueue: return MinimalQueue(node.get("queueName"), queue_keys, self.redisConnection, self.scripts, {"prefix": prefix}) - async def addChildren(self, nodes, parent, queues_opts, pipe): + async def addChildren(self, nodes: list[dict], parent: dict, queues_opts: Optional[dict], pipe: Any) -> list[dict]: children = [] for node in nodes: job = await self.addNode(node, parent, queues_opts, pipe) children.append(job) return children - async def addNodes(self, nodes: list[dict], pipe): + async def addNodes(self, nodes: list[dict], pipe: Any) -> list[dict]: trees = [] for node in nodes: parent_opts = node.get("opts", {}).get("parent", None) @@ -62,7 +62,7 @@ async def addNodes(self, nodes: list[dict], pipe): return trees - async def addNode(self, node: dict, parent: dict, queues_opts: dict, pipe): + async def addNode(self, node: dict, parent: dict, queues_opts: Optional[dict], pipe: Any) -> dict: prefix = node.get("prefix", self.prefix) queue = self.queueFromNode(node, QueueKeys(prefix), prefix) queue_name = node.get("queueName") @@ -116,7 +116,7 @@ async def addNode(self, node: dict, parent: dict, queues_opts: dict, pipe): return {"job": job} - async def add(self, flow: dict, opts: dict = {}): + async def add(self, flow: dict, opts: dict = {}) -> dict: parent_opts = flow.get("opts", {}).get("parent", None) result = None @@ -127,7 +127,7 @@ async def add(self, flow: dict, opts: dict = {}): return result - async def addBulk(self, flows: list[dict]): + async def addBulk(self, flows: list[dict]) -> list[dict]: result = None async with self.redisConnection.conn.pipeline(transaction=True) as pipe: job_trees = await self.addNodes(flows, pipe) From cc5b767cc6a8a3031248864de4bf998b7f4d1b3f Mon Sep 17 00:00:00 2001 From: yogeshwaran-c Date: Wed, 22 Apr 2026 11:17:09 +0530 Subject: [PATCH 2/3] fix(python): avoid mutable default and fix result type in FlowProducer.add per review --- python/bullmq/flow_producer.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/python/bullmq/flow_producer.py b/python/bullmq/flow_producer.py index d56412f6348..05cbe2f93af 100644 --- a/python/bullmq/flow_producer.py +++ b/python/bullmq/flow_producer.py @@ -116,16 +116,14 @@ async def addNode(self, node: dict, parent: dict, queues_opts: Optional[dict], p return {"job": job} - async def add(self, flow: dict, opts: dict = {}) -> dict: + async def add(self, flow: dict, opts: Optional[dict] = None) -> dict: + opts = opts if opts is not None else {} parent_opts = flow.get("opts", {}).get("parent", None) - result = None async with self.redisConnection.conn.pipeline(transaction=True) as pipe: jobs_tree = await self.addNode(flow, {"parentOpts": parent_opts},opts.get("queuesOptions"), pipe) await pipe.execute() - result = jobs_tree - - return result + return jobs_tree async def addBulk(self, flows: list[dict]) -> list[dict]: result = None From 12a0b11e20a38c417a74ec9e1b90f09b4f21fd6e Mon Sep 17 00:00:00 2001 From: yogeshwaran-c Date: Thu, 30 Apr 2026 16:28:44 +0530 Subject: [PATCH 3/3] fix(flow-producer): avoid mutable default args and Optional return drift [python] - Replace 'opts: dict = {}' with 'opts: dict | None = None' (PEP 604) to avoid mutable default - Tighten dict-returning methods so the return value is never inferred as Optional[dict] --- python/bullmq/flow_producer.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/python/bullmq/flow_producer.py b/python/bullmq/flow_producer.py index 05cbe2f93af..260b4afdf12 100644 --- a/python/bullmq/flow_producer.py +++ b/python/bullmq/flow_producer.py @@ -12,10 +12,11 @@ class MinimalQueue: Instantiate a MinimalQueue object """ - def __init__(self, name: str, queue_keys, redisConnection, scripts, opts: QueueBaseOptions = {}): + def __init__(self, name: str, queue_keys, redisConnection, scripts, opts: QueueBaseOptions | None = None): """ Initialize a connection """ + opts = opts or {} self.name = name self.redisConnection = redisConnection self.client = self.redisConnection.conn @@ -32,10 +33,13 @@ class FlowProducer: """ #TODO: pass only queueOpts, no need 2 parameters in next breaking change - def __init__(self, redisOpts: Union[dict, str] = {}, opts: QueueBaseOptions = {}): + def __init__(self, redisOpts: Union[dict, str] | None = None, opts: QueueBaseOptions | None = None): """ Initialize a connection """ + if redisOpts is None: + redisOpts = {} + opts = opts or {} self.redisConnection = RedisConnection(redisOpts) self.client = self.redisConnection.conn self.opts: dict = opts @@ -116,8 +120,8 @@ async def addNode(self, node: dict, parent: dict, queues_opts: Optional[dict], p return {"job": job} - async def add(self, flow: dict, opts: Optional[dict] = None) -> dict: - opts = opts if opts is not None else {} + async def add(self, flow: dict, opts: dict | None = None) -> dict: + opts = opts or {} parent_opts = flow.get("opts", {}).get("parent", None) async with self.redisConnection.conn.pipeline(transaction=True) as pipe: @@ -126,13 +130,10 @@ async def add(self, flow: dict, opts: Optional[dict] = None) -> dict: return jobs_tree async def addBulk(self, flows: list[dict]) -> list[dict]: - result = None async with self.redisConnection.conn.pipeline(transaction=True) as pipe: job_trees = await self.addNodes(flows, pipe) await pipe.execute() - result = job_trees - - return result + return job_trees async def close(self): """