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-2026 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 

22if typing.TYPE_CHECKING: 

23 from .base import Engine 

24 from .interfaces import _CoreAnyExecuteParams 

25 from .interfaces import CoreExecuteOptionsParameter 

26 from .interfaces import Dialect 

27 from .url import URL 

28 from ..sql.base import Executable 

29 from ..sql.ddl import InvokeDDLBase 

30 from ..sql.schema import HasSchemaAttr 

31 from ..sql.visitors import Visitable 

32 

33 

34class MockConnection: 

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

36 self._dialect = dialect 

37 self._execute_impl = execute 

38 

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

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

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

42 

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

44 return self 

45 

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

47 return obj.schema 

48 

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

50 return self 

51 

52 def _run_ddl_visitor( 

53 self, 

54 visitorcallable: Type[InvokeDDLBase], 

55 element: Visitable, 

56 **kwargs: Any, 

57 ) -> None: 

58 kwargs["checkfirst"] = False 

59 visitorcallable( 

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

61 ).traverse_single(element) 

62 

63 def execute( 

64 self, 

65 obj: Executable, 

66 parameters: Optional[_CoreAnyExecuteParams] = None, 

67 execution_options: Optional[CoreExecuteOptionsParameter] = None, 

68 ) -> Any: 

69 return self._execute_impl(obj, parameters) 

70 

71 

72def create_mock_engine( 

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

74) -> MockConnection: 

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

76 

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

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

79 and related methods. 

80 

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

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

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

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

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

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

87 receiving any results. 

88 

89 E.g.:: 

90 

91 from sqlalchemy import create_mock_engine 

92 

93 

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

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

96 

97 

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

99 metadata.create_all(engine, checkfirst=False) 

100 

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

102 database backend name. 

103 

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

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

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

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

108 

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

110 the previous "mock" engine strategy used with 

111 :func:`_sa.create_engine`. 

112 

113 .. seealso:: 

114 

115 :ref:`faq_ddl_as_string` 

116 

117 """ 

118 

119 # create url.URL object 

120 u = _url.make_url(url) 

121 

122 dialect_cls = u.get_dialect() 

123 

124 dialect_args = {} 

125 # consume dialect arguments from kwargs 

126 for k in util.get_cls_kwargs(dialect_cls): 

127 if k in kw: 

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

129 

130 # create dialect 

131 dialect = dialect_cls(**dialect_args) 

132 

133 return MockConnection(dialect, executor)