1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
|
# engine/mock.py
# Copyright (C) 2005-2019 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
from operator import attrgetter
from . import base
from . import url as _url
from .. import util
from ..sql import schema
from ..sql import ddl
class MockConnection(base.Connectable):
def __init__(self, dialect, execute):
self._dialect = dialect
self.execute = execute
engine = property(lambda s: s)
dialect = property(attrgetter("_dialect"))
name = property(lambda s: s._dialect.name)
schema_for_object = schema._schema_getter(None)
def connect(self, **kwargs):
return self
def execution_options(self, **kw):
return self
def compiler(self, statement, parameters, **kwargs):
return self._dialect.compiler(
statement, parameters, engine=self, **kwargs
)
def create(self, entity, **kwargs):
kwargs["checkfirst"] = False
ddl.SchemaGenerator(self.dialect, self, **kwargs).traverse_single(
entity
)
def drop(self, entity, **kwargs):
kwargs["checkfirst"] = False
ddl.SchemaDropper(self.dialect, self, **kwargs).traverse_single(entity)
def _run_visitor(
self, visitorcallable, element, connection=None, **kwargs
):
kwargs["checkfirst"] = False
visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
def execute(self, object_, *multiparams, **params):
raise NotImplementedError()
def create_mock_engine(url, executor, **kw):
"""Create a "mock" engine used for echoing DDL.
.. versionadded:: 2.0
"""
# create url.URL object
u = _url.make_url(url)
dialect_cls = u.get_dialect()
dialect_args = {}
# consume dialect arguments from kwargs
for k in util.get_cls_kwargs(dialect_cls):
if k in kw:
dialect_args[k] = kwargs.pop(k)
# create dialect
dialect = dialect_cls(**dialect_args)
return MockConnection(dialect, executor)
|
