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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
# Copyright 2012-present MongoDB, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Motor specific extensions to Sphinx."""
import re
from docutils.nodes import doctest_block, literal_block
from sphinx import addnodes
from sphinx.addnodes import desc, desc_content, desc_signature, seealso, versionmodified
from sphinx.util.inspect import safe_getattr
import motor
import motor.core
# This is a place to store info while parsing, to be used before generating.
motor_info = {}
def has_node_of_type(root, klass):
if isinstance(root, klass):
return True
for child in root.children: # noqa: SIM110
if has_node_of_type(child, klass):
return True
return False
def find_by_path(root, classes):
if not classes:
return [root]
_class = classes[0]
rv = []
for child in root.children:
if isinstance(child, _class):
rv.extend(find_by_path(child, classes[1:]))
return rv
docstring_warnings = []
def maybe_warn_about_code_block(name, content_node):
if has_node_of_type(content_node, (literal_block, doctest_block)):
docstring_warnings.append(name)
def has_coro_annotation(signature_node):
try:
return "coroutine" in signature_node[0][0]
except IndexError:
return False
def process_motor_nodes(app, doctree):
# Search doctree for Motor's methods and attributes whose docstrings were
# copied from PyMongo, and fix them up for Motor:
# 1. Add a 'coroutine' annotation to the beginning of the declaration.
# 2. Remove all version annotations like "New in version 2.0" since
# PyMongo's version numbers are meaningless in Motor's docs.
# 3. Remove "seealso" directives that reference PyMongo's docs.
#
# We do this here, rather than by registering a callback to Sphinx's
# 'autodoc-process-signature' event, because it's way easier to handle the
# parsed doctree before it's turned into HTML than it is to update the RST.
for objnode in doctree.traverse(desc):
if objnode["objtype"] in ("method", "attribute"):
signature_node = find_by_path(objnode, [desc_signature])[0]
name = ".".join([signature_node["module"], signature_node["fullname"]])
assert name.startswith("motor.")
obj_motor_info = motor_info.get(name)
if obj_motor_info:
desc_content_node = find_by_path(objnode, [desc_content])[0]
if desc_content_node.line is None and obj_motor_info["is_pymongo_docstring"]:
maybe_warn_about_code_block(name, desc_content_node)
if obj_motor_info["is_async_method"]: # noqa: SIM102
# Might be a handwritten RST with "coroutine" already.
if not has_coro_annotation(signature_node):
coro_annotation = addnodes.desc_annotation(
"coroutine ", "coroutine ", classes=["coro-annotation"]
)
signature_node.insert(0, coro_annotation)
if obj_motor_info["is_pymongo_docstring"]:
# Remove all "versionadded", "versionchanged" and
# "deprecated" directives from the docs we imported from
# PyMongo
version_nodes = find_by_path(desc_content_node, [versionmodified])
for version_node in version_nodes:
version_node.parent.remove(version_node)
# Remove all "seealso" directives that contain :doc:
# references from PyMongo's docs
seealso_nodes = find_by_path(desc_content_node, [seealso])
for seealso_node in seealso_nodes:
if 'reftype="doc"' in str(seealso_node):
seealso_node.parent.remove(seealso_node)
def get_motor_attr(motor_class, name, *defargs):
"""If any Motor attributes can't be accessed, grab the equivalent PyMongo
attribute. While we're at it, store some info about each attribute
in the global motor_info dict.
"""
attr = safe_getattr(motor_class, name, *defargs)
# Store some info for process_motor_nodes()
full_name = f"{motor_class.__module__}.{motor_class.__name__}.{name}"
full_name_legacy = f"motor.{motor_class.__module__}.{motor_class.__name__}.{name}"
# These sub-attributes are set in motor.asynchronize()
has_coroutine_annotation = getattr(attr, "coroutine_annotation", False)
is_async_method = getattr(attr, "is_async_method", False)
is_cursor_method = getattr(attr, "is_motorcursor_chaining_method", False)
if is_async_method or is_cursor_method:
pymongo_method = getattr(motor_class.__delegate_class__, attr.pymongo_method_name)
else:
pymongo_method = None
# attr.doc is set by statement like 'error = AsyncRead(doc="OBSOLETE")'.
is_pymongo_doc = pymongo_method and attr.__doc__ == pymongo_method.__doc__
motor_info[full_name] = motor_info[full_name_legacy] = {
"is_async_method": is_async_method or has_coroutine_annotation,
"is_pymongo_docstring": is_pymongo_doc,
"pymongo_method": pymongo_method,
}
return attr
pymongo_ref_pat = re.compile(r":doc:`(.*?)`", re.MULTILINE)
def _sub_pymongo_ref(match):
ref = match.group(1)
return ":doc:`%s`" % ref.lstrip("/")
def process_motor_docstring(app, what, name, obj, options, lines):
if name in motor_info and motor_info[name].get("is_pymongo_docstring"):
joined = "\n".join(lines)
subbed = pymongo_ref_pat.sub(_sub_pymongo_ref, joined)
lines[:] = subbed.split("\n")
def build_finished(app, exception):
if not exception and docstring_warnings:
print("PyMongo docstrings with code blocks that need update:")
for name in sorted(docstring_warnings):
print(name)
def setup(app):
app.add_autodoc_attrgetter(type(motor.core.AgnosticBase), get_motor_attr)
app.connect("autodoc-process-docstring", process_motor_docstring)
app.connect("doctree-read", process_motor_nodes)
app.connect("build-finished", build_finished)
return {"parallel_write_safe": True, "parallel_read_safe": False}
|
