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

1import re as py_re 

2import typing 

3 

4from archdocs import diagram_model, settings 

5from archdocs.features.sqlalchemy.const import SQLAlchemyFeatures 

6 

7 

8_DSN_CREDENTIALS_PATTERN: typing.Final = py_re.compile(r"://[^/@\s]*@", flags=settings.TYPICAL_RE_FLAGS) 

9_DSN_SCHEME_PATTERN: typing.Final = py_re.compile(r"^[^:/\s]+", flags=settings.TYPICAL_RE_FLAGS) 

10 

11 

12def _build_database_node(raw_database_type: str, node_suffix: int | str) -> diagram_model.DiagramNode: 

13 scheme_match: typing.Final = _DSN_SCHEME_PATTERN.search(raw_database_type) 

14 database_scheme: typing.Final = scheme_match.group() if scheme_match else raw_database_type 

15 return diagram_model.build_diagram_node( 

16 f"{diagram_model.render_node_id(database_scheme)}db{node_suffix}", 

17 database_scheme if node_suffix == "" else f"{database_scheme} #{node_suffix}", 

18 diagram_model.NodeGroup.data_stores, 

19 ) 

20 

21 

22def render_sqlalchemy_features( 

23 service_node: diagram_model.DiagramNode, 

24 features_to_draw: SQLAlchemyFeatures, 

25 /, 

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

27 if not features_to_draw.database_type: 

28 return () 

29 drawn_dsn: typing.Final = _DSN_CREDENTIALS_PATTERN.sub("://***@", features_to_draw.database_type) 

30 # A dsn carrying the attribute in its query string already spells it out on the arrow, so the 

31 # separate mention is what the attribute set outside the dsn — `connect_args`, a setting — 

32 # would otherwise lose. 

33 session_attrs_to_draw: typing.Final = ( 

34 "" if features_to_draw.target_session_attrs in drawn_dsn else features_to_draw.target_session_attrs 

35 ) 

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

37 filter( 

38 None, 

39 [ 

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

41 drawn_dsn, 

42 session_attrs_to_draw, 

43 ], 

44 ), 

45 ) 

46 connections_number: typing.Final = ( 

47 settings.VALUE_FOR_MASS_CONNECTIONS_ILLUSTRATION if features_to_draw.pooling_used else 1 

48 ) 

49 return tuple( 

50 diagram_model.DiagramEdge( 

51 source_node=service_node, 

52 target_node=_build_database_node( 

53 features_to_draw.database_type, 

54 one_counter if features_to_draw.multiple_hosts else "", 

55 ), 

56 edge_label=properties_on_arrow, 

57 ) 

58 for one_counter in range(connections_number) 

59 )