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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
# event/legacy.py
# Copyright (C) 2005-2014 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
"""Routines to handle adaption of legacy call signatures,
generation of deprecation notes and docstrings.
"""
from .. import util
def _legacy_signature(since, argnames, converter=None):
def leg(fn):
if not hasattr(fn, '_legacy_signatures'):
fn._legacy_signatures = []
fn._legacy_signatures.append((since, argnames, converter))
return fn
return leg
def _wrap_fn_for_legacy(dispatch_descriptor, fn, argspec):
for since, argnames, conv in dispatch_descriptor.legacy_signatures:
if argnames[-1] == "**kw":
has_kw = True
argnames = argnames[0:-1]
else:
has_kw = False
if len(argnames) == len(argspec.args) \
and has_kw is bool(argspec.keywords):
if conv:
assert not has_kw
def wrap_leg(*args):
return fn(*conv(*args))
else:
def wrap_leg(*args, **kw):
argdict = dict(zip(dispatch_descriptor.arg_names, args))
args = [argdict[name] for name in argnames]
if has_kw:
return fn(*args, **kw)
else:
return fn(*args)
return wrap_leg
else:
return fn
def _indent(text, indent):
return "\n".join(
indent + line
for line in text.split("\n")
)
def _standard_listen_example(dispatch_descriptor, sample_target, fn):
example_kw_arg = _indent(
"\n".join(
"%(arg)s = kw['%(arg)s']" % {"arg": arg}
for arg in dispatch_descriptor.arg_names[0:2]
),
" ")
if dispatch_descriptor.legacy_signatures:
current_since = max(since for since, args, conv
in dispatch_descriptor.legacy_signatures)
else:
current_since = None
text = (
"from sqlalchemy import event\n\n"
"# standard decorator style%(current_since)s\n"
"@event.listens_for(%(sample_target)s, '%(event_name)s')\n"
"def receive_%(event_name)s(%(named_event_arguments)s%(has_kw_arguments)s):\n"
" \"listen for the '%(event_name)s' event\"\n"
"\n # ... (event handling logic) ...\n"
)
if len(dispatch_descriptor.arg_names) > 3:
text += (
"\n# named argument style (new in 0.9)\n"
"@event.listens_for(%(sample_target)s, '%(event_name)s', named=True)\n"
"def receive_%(event_name)s(**kw):\n"
" \"listen for the '%(event_name)s' event\"\n"
"%(example_kw_arg)s\n"
"\n # ... (event handling logic) ...\n"
)
text %= {
"current_since": " (arguments as of %s)" %
current_since if current_since else "",
"event_name": fn.__name__,
"has_kw_arguments": ", **kw" if dispatch_descriptor.has_kw else "",
"named_event_arguments": ", ".join(dispatch_descriptor.arg_names),
"example_kw_arg": example_kw_arg,
"sample_target": sample_target
}
return text
def _legacy_listen_examples(dispatch_descriptor, sample_target, fn):
text = ""
for since, args, conv in dispatch_descriptor.legacy_signatures:
text += (
"\n# legacy calling style (pre-%(since)s)\n"
"@event.listens_for(%(sample_target)s, '%(event_name)s')\n"
"def receive_%(event_name)s(%(named_event_arguments)s%(has_kw_arguments)s):\n"
" \"listen for the '%(event_name)s' event\"\n"
"\n # ... (event handling logic) ...\n" % {
"since": since,
"event_name": fn.__name__,
"has_kw_arguments": " **kw" if dispatch_descriptor.has_kw else "",
"named_event_arguments": ", ".join(args),
"sample_target": sample_target
}
)
return text
def _version_signature_changes(dispatch_descriptor):
since, args, conv = dispatch_descriptor.legacy_signatures[0]
return (
"\n.. versionchanged:: %(since)s\n"
" The ``%(event_name)s`` event now accepts the \n"
" arguments ``%(named_event_arguments)s%(has_kw_arguments)s``.\n"
" Listener functions which accept the previous argument \n"
" signature(s) listed above will be automatically \n"
" adapted to the new signature." % {
"since": since,
"event_name": dispatch_descriptor.__name__,
"named_event_arguments": ", ".join(dispatch_descriptor.arg_names),
"has_kw_arguments": ", **kw" if dispatch_descriptor.has_kw else ""
}
)
def _augment_fn_docs(dispatch_descriptor, parent_dispatch_cls, fn):
header = ".. container:: event_signatures\n\n"\
" Example argument forms::\n"\
"\n"
sample_target = getattr(parent_dispatch_cls, "_target_class_doc", "obj")
text = (
header +
_indent(
_standard_listen_example(
dispatch_descriptor, sample_target, fn),
" " * 8)
)
if dispatch_descriptor.legacy_signatures:
text += _indent(
_legacy_listen_examples(
dispatch_descriptor, sample_target, fn),
" " * 8)
text += _version_signature_changes(dispatch_descriptor)
return util.inject_docstring_text(fn.__doc__,
text,
1
)
|
