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, host_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{host_suffix}", 

17 database_scheme if host_suffix == "" else f"{database_scheme} #{host_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 properties_on_arrow: typing.Final = ", ".join( 

30 filter( 

31 None, 

32 [ 

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

34 _DSN_CREDENTIALS_PATTERN.sub("://***@", features_to_draw.database_type), 

35 features_to_draw.target_session_attrs, 

36 ], 

37 ), 

38 ) 

39 connections_number: typing.Final = ( 

40 settings.VALUE_FOR_MASS_CONNECTIONS_ILLUSTRATION if features_to_draw.pooling_used else 1 

41 ) 

42 return tuple( 

43 diagram_model.DiagramEdge( 

44 source_node=service_node, 

45 target_node=_build_database_node( 

46 features_to_draw.database_type, 

47 one_counter if features_to_draw.multiple_hosts else "", 

48 ), 

49 edge_label=properties_on_arrow, 

50 ) 

51 for one_counter in range(connections_number) 

52 )