chore: generate_code_snippets.py script to work with v8 async client#2677
chore: generate_code_snippets.py script to work with v8 async client#2677haakonvt wants to merge 2 commits into
generate_code_snippets.py script to work with v8 async client#2677Conversation
There was a problem hiding this comment.
Code Review
This pull request updates the code snippet generation script to support asynchronous clients by transitioning to AsyncCogniteClient and implementing custom API and method traversal. The review feedback recommends using inspect.getattr_static to make the traversal more robust against __slots__ or lazy-loaded attributes and to simplify the custom MRO iteration.
| import json | ||
| import re | ||
| from collections import defaultdict | ||
| from doctest import DocTestParser, Example | ||
| from functools import cached_property | ||
|
|
||
| from cognite.client import ClientConfig, CogniteClient | ||
| from cognite.client._api_client import APIClient | ||
| from cognite.client import ClientConfig | ||
| from cognite.client._basic_api_client import BasicAsyncAPIClient | ||
| from cognite.client._cognite_client import AsyncCogniteClient | ||
| from cognite.client.credentials import Token | ||
|
|
||
| _SKIP_DESCRIPTOR_TYPES = (property, cached_property, staticmethod, classmethod) | ||
|
|
||
|
|
||
| def collect_apis(obj, done): | ||
| if done.get(obj.__class__): | ||
| return [] | ||
| done[obj.__class__] = True | ||
| apis = inspect.getmembers(obj, lambda m: isinstance(m, APIClient) and not done.get(m.__class__)) | ||
| apis = [(n, v) for n, v in vars(obj).items() if isinstance(v, BasicAsyncAPIClient) and not done.get(v.__class__)] | ||
| sub = [(n + "." + sn, sa) for n, c in apis for sn, sa in collect_apis(c, done)] | ||
| return apis + sub | ||
|
|
||
|
|
||
| client = CogniteClient(ClientConfig(project="_", client_name="_", cluster="_", credentials=Token("_"))) | ||
| def iter_methods(api): | ||
| """Iterate bound methods without triggering class-level properties.""" | ||
| seen = set() | ||
| for cls in type(api).__mro__: | ||
| for name, val in cls.__dict__.items(): | ||
| if name in seen: | ||
| continue | ||
| seen.add(name) | ||
| if isinstance(val, _SKIP_DESCRIPTOR_TYPES) or not callable(val): | ||
| continue | ||
| yield name, getattr(api, name) |
There was a problem hiding this comment.
Use inspect.getattr_static for robust API and method traversal
Using vars(obj) in collect_apis is fragile because it only inspects instance attributes (__dict__). This will fail if any API or sub-API is defined using __slots__, or if they are lazily loaded via properties/descriptors in the future. Additionally, the manual MRO traversal in iter_methods is complex and can be simplified.
We can use inspect.getattr_static to safely inspect attributes and methods without triggering descriptors (like properties or cached properties), while maintaining full compatibility with __slots__ and lazy-loaded attributes.
import inspect
import json
import re
from collections import defaultdict
from doctest import DocTestParser, Example
from functools import cached_property
from cognite.client import ClientConfig
from cognite.client._basic_api_client import BasicAsyncAPIClient
from cognite.client._cognite_client import AsyncCogniteClient
from cognite.client.credentials import Token
_SKIP_DESCRIPTOR_TYPES = (property, cached_property, staticmethod, classmethod)
def collect_apis(obj, done):
if done.get(obj.__class__):
return []
done[obj.__class__] = True
apis = []
for name in dir(obj):
if name.startswith("_") or name == "api_client":
continue
try:
static_val = inspect.getattr_static(obj, name)
except AttributeError:
continue
if isinstance(static_val, _SKIP_DESCRIPTOR_TYPES):
continue
try:
val = getattr(obj, name)
except Exception:
continue
if isinstance(val, BasicAsyncAPIClient) and not done.get(val.__class__):
apis.append((name, val))
sub = [(n + "." + sn, sa) for n, c in apis for sn, sa in collect_apis(c, done)]
return apis + sub
def iter_methods(api):
"""Iterate bound methods without triggering class-level properties."""
for name in dir(api):
if name.startswith("_"):
continue
try:
static_val = inspect.getattr_static(api, name)
except AttributeError:
continue
if isinstance(static_val, _SKIP_DESCRIPTOR_TYPES):
continue
try:
member = getattr(api, name)
except Exception:
continue
if callable(member):
yield name, memberThere was a problem hiding this comment.
@gemini-code-assist Good catch on iter_methods, simplified it with dir() + getattr_static. Kept vars(obj) in collect_apis though: getattr_static on slotted or lazily-loaded attributes returns the descriptor, not the value, so it would silently miss APIs there.
There was a problem hiding this comment.
That is a fair point regarding vars(obj). If getattr_static returns the descriptor for slotted or lazily-loaded attributes, using it directly in collect_apis would indeed cause the script to miss those APIs. Your approach of using vars(obj) for discovery while using getattr_static for method iteration in iter_methods strikes the right balance between robustness and correctness for this specific use case.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #2677 +/- ##
==========================================
- Coverage 93.65% 93.64% -0.01%
==========================================
Files 498 498
Lines 50342 50342
==========================================
- Hits 47148 47144 -4
- Misses 3194 3198 +4 🚀 New features to boost your workflow:
|
|
🦄 |
The official documentation lags very far behind the SDK on Python examples. The script to create the links between "links" and "examples" does not even work after v8 so fix that first 😄
Added
iter_methods+_SKIP_DESCRIPTOR_TYPESto avoid the nasty surprise that theOrgAPIhas - cached_property making API calls....