Coverage for archdocs/features/redis/renderer.py: 100%

1import typing 

2 

3from archdocs import diagram_model, settings 

4from archdocs.features.redis.const import RedisFeatures 

5 

6 

7_REDIS_NODE_LABEL: typing.Final = "redis" 

8 

9 

10def _build_redis_node(connection_type: str, node_suffix: int | str, /) -> diagram_model.DiagramNode: 

11 node_kind: typing.Final = f"{_REDIS_NODE_LABEL} {connection_type}" if connection_type else _REDIS_NODE_LABEL 

12 return diagram_model.build_diagram_node( 

13 f"redisdb_{connection_type}{node_suffix}" if connection_type else "redisdb", 

14 node_kind if node_suffix == "" else f"{node_kind} #{node_suffix}", 

15 diagram_model.NodeGroup.data_stores, 

16 ) 

17 

18 

19def render_redis_features( 

20 service_node: diagram_model.DiagramNode, 

21 features_to_draw: RedisFeatures, 

22 /, 

23) -> tuple[diagram_model.DiagramEdge, ...]: 

24 if not features_to_draw.connection_type and not features_to_draw.async_used and not features_to_draw.retry_used: 

25 return () 

26 properties_on_arrow: typing.Final = ", ".join( 

27 filter( 

28 None, 

29 [ 

30 "async" if features_to_draw.async_used else "", 

31 "retry" if features_to_draw.retry_used else "", 

32 ], 

33 ), 

34 ) 

35 if not features_to_draw.cluster_or_sentinel: 

36 return ( 

37 diagram_model.DiagramEdge( 

38 source_node=service_node, 

39 target_node=_build_redis_node("", ""), 

40 edge_label=properties_on_arrow, 

41 ), 

42 ) 

43 return tuple( 

44 diagram_model.DiagramEdge( 

45 source_node=service_node, 

46 target_node=_build_redis_node(features_to_draw.connection_type or "", one_counter), 

47 edge_label=properties_on_arrow, 

48 ) 

49 for one_counter in range(settings.VALUE_FOR_MASS_CONNECTIONS_ILLUSTRATION) 

50 )