Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/mock.py: 47%

1# engine/mock.py 

2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: https://www.opensource.org/licenses/mit-license.php 

7 

8from __future__ import annotations 

9 

10from operator import attrgetter 

11import typing 

12from typing import Any 

13from typing import Callable 

14from typing import cast 

15from typing import Optional 

16from typing import Type 

17from typing import Union 

18 

19from . import url as _url 

20from .. import util 

21 

22 

23if typing.TYPE_CHECKING: 

24 from .base import Engine 

25 from .interfaces import _CoreAnyExecuteParams 

26 from .interfaces import CoreExecuteOptionsParameter 

27 from .interfaces import Dialect 

28 from .url import URL 

29 from ..sql.base import Executable 

30 from ..sql.ddl import InvokeDDLBase 

31 from ..sql.schema import HasSchemaAttr 

32 from ..sql.visitors import Visitable 

33 

34 

35class MockConnection: 

36 def __init__(self, dialect: Dialect, execute: Callable[..., Any]): 

37 self._dialect = dialect 

38 self._execute_impl = execute 

39 

40 engine: Engine = cast(Any, property(lambda s: s)) 

41 dialect: Dialect = cast(Any, property(attrgetter("_dialect"))) 

42 name: str = cast(Any, property(lambda s: s._dialect.name)) 

43 

44 def connect(self, **kwargs: Any) -> MockConnection: 

45 return self 

46 

47 def schema_for_object(self, obj: HasSchemaAttr) -> Optional[str]: 

48 return obj.schema 

49 

50 def execution_options(self, **kw: Any) -> MockConnection: 

51 return self 

52 

53 def _run_ddl_visitor( 

54 self, 

55 visitorcallable: Type[InvokeDDLBase], 

56 element: Visitable, 

57 **kwargs: Any, 

58 ) -> None: 

59 kwargs["checkfirst"] = False 

60 visitorcallable( 

61 dialect=self.dialect, connection=self, **kwargs 

62 ).traverse_single(element) 

63 

64 def execute( 

65 self, 

66 obj: Executable, 

67 parameters: Optional[_CoreAnyExecuteParams] = None, 

68 execution_options: Optional[CoreExecuteOptionsParameter] = None, 

69 ) -> Any: 

70 return self._execute_impl(obj, parameters) 

71 

72 

73def create_mock_engine( 

74 url: Union[str, URL], executor: Any, **kw: Any 

75) -> MockConnection: 

76 """Create a "mock" engine used for echoing DDL. 

77 

78 This is a utility function used for debugging or storing the output of DDL 

79 sequences as generated by :meth:`_schema.MetaData.create_all` 

80 and related methods. 

81 

82 The function accepts a URL which is used only to determine the kind of 

83 dialect to be used, as well as an "executor" callable function which 

84 will receive a SQL expression object and parameters, which can then be 

85 echoed or otherwise printed. The executor's return value is not handled, 

86 nor does the engine allow regular string statements to be invoked, and 

87 is therefore only useful for DDL that is sent to the database without 

88 receiving any results. 

89 

90 E.g.:: 

91 

92 from sqlalchemy import create_mock_engine 

93 

94 

95 def dump(sql, *multiparams, **params): 

96 print(sql.compile(dialect=engine.dialect)) 

97 

98 

99 engine = create_mock_engine("postgresql+psycopg2://", dump) 

100 metadata.create_all(engine, checkfirst=False) 

101 

102 :param url: A string URL which typically needs to contain only the 

103 database backend name. 

104 

105 :param executor: a callable which receives the arguments ``sql``, 

106 ``*multiparams`` and ``**params``. The ``sql`` parameter is typically 

107 an instance of :class:`.ExecutableDDLElement`, which can then be compiled 

108 into a string using :meth:`.ExecutableDDLElement.compile`. 

109 

110 .. versionadded:: 1.4 - the :func:`.create_mock_engine` function replaces 

111 the previous "mock" engine strategy used with 

112 :func:`_sa.create_engine`. 

113 

114 .. seealso:: 

115 

116 :ref:`faq_ddl_as_string` 

117 

118 """ 

119 

120 # create url.URL object 

121 u = _url.make_url(url) 

122 

123 dialect_cls = u.get_dialect() 

124 

125 dialect_args = {} 

126 # consume dialect arguments from kwargs 

127 for k in util.get_cls_kwargs(dialect_cls): 

128 if k in kw: 

129 dialect_args[k] = kw.pop(k) 

130 

131 # create dialect 

132 dialect = dialect_cls(**dialect_args) 

133 

134 return MockConnection(dialect, executor)