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

1# engine/mock.py 

2# Copyright (C) 2005-2024 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 operator import attrgetter 

9 

10from . import base 

11from . import url as _url 

12from .. import util 

13from ..sql import ddl 

14 

15 

16class MockConnection(base.Connectable): 

17 def __init__(self, dialect, execute): 

18 self._dialect = dialect 

19 self.execute = execute 

20 

21 engine = property(lambda s: s) 

22 dialect = property(attrgetter("_dialect")) 

23 name = property(lambda s: s._dialect.name) 

24 

25 def schema_for_object(self, obj): 

26 return obj.schema 

27 

28 def connect(self, **kwargs): 

29 return self 

30 

31 def execution_options(self, **kw): 

32 return self 

33 

34 def compiler(self, statement, parameters, **kwargs): 

35 return self._dialect.compiler( 

36 statement, parameters, engine=self, **kwargs 

37 ) 

38 

39 def create(self, entity, **kwargs): 

40 kwargs["checkfirst"] = False 

41 

42 ddl.SchemaGenerator(self.dialect, self, **kwargs).traverse_single( 

43 entity 

44 ) 

45 

46 def drop(self, entity, **kwargs): 

47 kwargs["checkfirst"] = False 

48 

49 ddl.SchemaDropper(self.dialect, self, **kwargs).traverse_single(entity) 

50 

51 def _run_ddl_visitor( 

52 self, visitorcallable, element, connection=None, **kwargs 

53 ): 

54 kwargs["checkfirst"] = False 

55 visitorcallable(self.dialect, self, **kwargs).traverse_single(element) 

56 

57 def execute(self, object_, *multiparams, **params): 

58 raise NotImplementedError() 

59 

60 

61def create_mock_engine(url, executor, **kw): 

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

63 

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

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

66 and related methods. 

67 

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

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

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

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

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

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

74 receiving any results. 

75 

76 E.g.:: 

77 

78 from sqlalchemy import create_mock_engine 

79 

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

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

82 

83 engine = create_mock_engine('postgresql://', dump) 

84 metadata.create_all(engine, checkfirst=False) 

85 

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

87 database backend name. 

88 

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

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

91 an instance of :class:`.DDLElement`, which can then be compiled into a 

92 string using :meth:`.DDLElement.compile`. 

93 

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

95 the previous "mock" engine strategy used with 

96 :func:`_sa.create_engine`. 

97 

98 .. seealso:: 

99 

100 :ref:`faq_ddl_as_string` 

101 

102 """ 

103 

104 # create url.URL object 

105 u = _url.make_url(url) 

106 

107 dialect_cls = u.get_dialect() 

108 

109 dialect_args = {} 

110 # consume dialect arguments from kwargs 

111 for k in util.get_cls_kwargs(dialect_cls): 

112 if k in kw: 

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

114 

115 # create dialect 

116 dialect = dialect_cls(**dialect_args) 

117 

118 return MockConnection(dialect, executor)