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# The windows are what keep both patterns honest on somebody's whole project: an unbounded gap 

10# reads "pool_ anywhere later in the file" as pooling and costs quadratic time on sources where 

11# `create_engine(` repeats without a match, a killer measured in minutes on a single file. 

12_POOLING_PATTERN: typing.Final = py_re.compile(r"create_engine\(.{0,2000}?pool_", settings.TYPICAL_RE_FLAGS) 

13_MULTIPLE_HOSTS_PATTERN: typing.Final = py_re.compile( 

14 r"create_engine\([^)'\"]{0,200}['\"](?:postgresql|mysql|mariadb|oracle|mssql)\+[^'\"]*://[^/,]+(?:,[^/,]+)+/", 

15 settings.TYPICAL_RE_FLAGS, 

16) 

17# The attribute is written three ways and the value is hyphenated in the two that matter: 

18# `?target_session_attrs=read-write` inside a dsn, `"target_session_attrs": "read-write"` in 

19# `connect_args`, and a bare setting. Quotes around the key and around the value are optional 

20# on purpose — demanding them is what used to make this pattern unable to match a dsn at all. 

21_TARGET_SESSION_ATTRS_PATTERN: typing.Final = py_re.compile( 

22 r"target_session_attrs['\"]?\s*[=:]\s*['\"]?(?P<attrs_value>[\w-]+)", 

23 settings.TYPICAL_RE_FLAGS, 

24) 

25_DB_TYPE_PATTERN: typing.Final = py_re.compile( 

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

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

28 settings.TYPICAL_RE_FLAGS, 

29) 

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

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

32_SQLALCHEMY_LITERALS: typing.Final = ( 

33 "sqlalchemy", 

34 "create_engine", 

35 "target_session_attrs", 

36 "postgresql", 

37 "mysql", 

38 "sqlite", 

39 "oracle", 

40 "mssql", 

41 "mariadb", 

42 "cockroachdb", 

43) 

44_EMPTY_FEATURES: typing.Final = SQLAlchemyFeatures( 

45 async_used=False, 

46 pooling_used=False, 

47 multiple_hosts=False, 

48 target_session_attrs="", 

49 database_type="", 

50) 

51 

52 

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

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

55 return _EMPTY_FEATURES 

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

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

58 return SQLAlchemyFeatures( 

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

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

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

62 target_session_attrs=target_session_attrs_match.group("attrs_value") if target_session_attrs_match else "", 

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

64 )