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

1import re as py_re 

2import typing 

3 

4from archdocs import prefilter, settings 

5from archdocs.features.sqlalchemy.const import SQLAlchemyFeatures 

6 

7 

8_ASYNC_ENGINE_PATTERN: typing.Final = py_re.compile(r"sqlalchemy\.ext\.asyncio", settings.TYPICAL_RE_FLAGS) 

9_POOLING_PATTERN: typing.Final = py_re.compile(r"create_engine\(.+pool_", settings.TYPICAL_RE_FLAGS) 

10_MULTIPLE_HOSTS_PATTERN: typing.Final = py_re.compile( 

11 r"create_engine\([^)]*'(?:postgresql|mysql|mariadb|oracle|mssql)\+[^']*://[^/]+,[^/]+/", 

12 settings.TYPICAL_RE_FLAGS, 

13) 

14_TARGET_SESSION_ATTRS_PATTERN: typing.Final = py_re.compile( 

15 r"target_session_attrs\s*=\s*['\"](\w+)['\"]", 

16 settings.TYPICAL_RE_FLAGS, 

17) 

18_DB_TYPE_PATTERN: typing.Final = py_re.compile( 

19 r"['\"](postgresql(?:\+[^'\"]*)?|mysql(?:\+[^'\"]*)?|sqlite(?:\+[^'\"]*)?|oracle(?:\+[^'\"]*)?|" 

20 r"mssql(?:\+[^'\"]*)?|mariadb(?:\+[^'\"]*)?|cockroachdb(?:\+[^'\"]*)?)['\"]", 

21 settings.TYPICAL_RE_FLAGS, 

22) 

23# One union of literals for the whole feature: an engine is found not only by the package 

24# name but by a bare DSN in a string, so the database schemes of `_DB_TYPE_PATTERN` belong here too. 

25_SQLALCHEMY_LITERALS: typing.Final = ( 

26 "sqlalchemy", 

27 "create_engine", 

28 "target_session_attrs", 

29 "postgresql", 

30 "mysql", 

31 "sqlite", 

32 "oracle", 

33 "mssql", 

34 "mariadb", 

35 "cockroachdb", 

36) 

37_EMPTY_FEATURES: typing.Final = SQLAlchemyFeatures( 

38 async_used=False, 

39 pooling_used=False, 

40 multiple_hosts=False, 

41 target_session_attrs="", 

42 database_type="", 

43) 

44 

45 

46def find_sqlalchemy_features(raw_source: str) -> SQLAlchemyFeatures: 

47 if not prefilter.contains_any_literal(raw_source.lower(), _SQLALCHEMY_LITERALS): 

48 return _EMPTY_FEATURES 

49 target_session_attrs_match: typing.Final = _TARGET_SESSION_ATTRS_PATTERN.search(raw_source) 

50 database_type_match: typing.Final = _DB_TYPE_PATTERN.search(raw_source) 

51 return SQLAlchemyFeatures( 

52 async_used=_ASYNC_ENGINE_PATTERN.search(raw_source) is not None, 

53 pooling_used=_POOLING_PATTERN.search(raw_source) is not None, 

54 multiple_hosts=_MULTIPLE_HOSTS_PATTERN.search(raw_source) is not None, 

55 target_session_attrs=target_session_attrs_match.group(1) if target_session_attrs_match else "", 

56 database_type=database_type_match.group(1) if database_type_match else "", 

57 )