codekingpro/portable-devtools

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
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)
135