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)