0
# Schema Introspection
1
2
Schema introspection capabilities for examining database structure, type information, and query metadata.
3
4
## Capabilities
5
6
### Type Description Classes
7
8
Classes for representing EdgeDB type information and schema elements.
9
10
```python { .api }
11
class AnyType:
12
"""
13
Base type descriptor for EdgeDB types.
14
15
Provides common attributes for all EdgeDB type descriptors.
16
"""
17
18
def __init__(self, desc_id: UUID, name: Optional[str] = None):
19
"""
20
Create type descriptor.
21
22
Parameters:
23
- desc_id: Unique identifier for the type
24
- name: Optional type name
25
"""
26
27
@property
28
def desc_id(self) -> UUID:
29
"""Unique identifier for this type."""
30
31
@property
32
def name(self) -> Optional[str]:
33
"""Type name if available."""
34
35
class Element:
36
"""
37
Schema element descriptor.
38
39
Represents a property, link, or link property in the schema.
40
"""
41
42
def __init__(
43
self,
44
type: AnyType,
45
cardinality: Cardinality,
46
is_implicit: bool = False,
47
kind: ElementKind = ElementKind.PROPERTY
48
):
49
"""
50
Create element descriptor.
51
52
Parameters:
53
- type: Element type descriptor
54
- cardinality: Element cardinality
55
- is_implicit: Whether element is implicit
56
- kind: Kind of schema element
57
"""
58
59
@property
60
def type(self) -> AnyType:
61
"""Element type."""
62
63
@property
64
def cardinality(self) -> Cardinality:
65
"""Element cardinality."""
66
67
@property
68
def is_implicit(self) -> bool:
69
"""Whether element is implicit."""
70
71
@property
72
def kind(self) -> ElementKind:
73
"""Schema element kind."""
74
75
class SequenceType(AnyType):
76
"""
77
Base descriptor for sequence types (arrays, sets, etc.).
78
79
Extends AnyType with element type information.
80
"""
81
82
def __init__(self, desc_id: UUID, name: Optional[str], element_type: AnyType):
83
"""
84
Create sequence type descriptor.
85
86
Parameters:
87
- desc_id: Type identifier
88
- name: Type name
89
- element_type: Type of sequence elements
90
"""
91
92
@property
93
def element_type(self) -> AnyType:
94
"""Type of sequence elements."""
95
96
class SetType(SequenceType):
97
"""
98
Set type descriptor.
99
100
Represents EdgeDB set types.
101
"""
102
```
103
104
### Introspection Context and Results
105
106
Context and result classes for schema introspection operations.
107
108
```python { .api }
109
class DescribeContext:
110
"""
111
Context for schema introspection operations.
112
113
Controls what information to include in introspection results.
114
"""
115
116
def __init__(
117
self,
118
query: str,
119
state: Optional[State] = None,
120
inject_type_names: bool = False,
121
output_format: OutputFormat = OutputFormat.BINARY,
122
expect_one: bool = False
123
):
124
"""
125
Create describe context.
126
127
Parameters:
128
- query: Query to introspect
129
- state: Client state for introspection
130
- inject_type_names: Whether to inject type names
131
- output_format: Output format for introspection
132
- expect_one: Whether to expect single result
133
"""
134
135
@property
136
def query(self) -> str:
137
"""Query being introspected."""
138
139
@property
140
def state(self) -> Optional[State]:
141
"""Client state for introspection."""
142
143
@property
144
def inject_type_names(self) -> bool:
145
"""Whether to inject type names."""
146
147
@property
148
def output_format(self) -> OutputFormat:
149
"""Output format for results."""
150
151
@property
152
def expect_one(self) -> bool:
153
"""Whether single result expected."""
154
155
class DescribeResult:
156
"""
157
Result of schema introspection operation.
158
159
Contains type information and metadata about a query.
160
"""
161
162
def __init__(
163
self,
164
input_type: Optional[AnyType] = None,
165
output_type: Optional[AnyType] = None,
166
output_cardinality: Cardinality = Cardinality.MANY,
167
capabilities: int = 0
168
):
169
"""
170
Create describe result.
171
172
Parameters:
173
- input_type: Input parameter type descriptor
174
- output_type: Output result type descriptor
175
- output_cardinality: Output cardinality
176
- capabilities: Required capabilities for query
177
"""
178
179
@property
180
def input_type(self) -> Optional[AnyType]:
181
"""Input parameter type."""
182
183
@property
184
def output_type(self) -> Optional[AnyType]:
185
"""Output result type."""
186
187
@property
188
def output_cardinality(self) -> Cardinality:
189
"""Output cardinality."""
190
191
@property
192
def capabilities(self) -> int:
193
"""Required capabilities bitmask."""
194
```
195
196
### Introspection Functions
197
198
Functions for examining database schema and query structure.
199
200
```python { .api }
201
def introspect_type(client: Union[Client, AsyncIOClient], type_name: str) -> AnyType:
202
"""
203
Introspect a specific type in the database schema.
204
205
Parameters:
206
- client: EdgeDB client instance
207
- type_name: Name of type to introspect
208
209
Returns:
210
Type descriptor for the specified type
211
"""
212
213
def describe_query(
214
client: Union[Client, AsyncIOClient],
215
query: str,
216
*args,
217
**kwargs
218
) -> DescribeResult:
219
"""
220
Describe a query's input and output types.
221
222
Parameters:
223
- client: EdgeDB client instance
224
- query: EdgeQL query to describe
225
- *args: Query arguments
226
- **kwargs: Named query arguments
227
228
Returns:
229
Description of query types and cardinality
230
"""
231
232
def get_schema_version(client: Union[Client, AsyncIOClient]) -> str:
233
"""
234
Get the current schema version.
235
236
Parameters:
237
- client: EdgeDB client instance
238
239
Returns:
240
Schema version string
241
"""
242
```
243
244
## Usage Examples
245
246
### Basic Type Introspection
247
248
```python
249
import edgedb
250
251
client = edgedb.create_client()
252
253
# Describe a simple query
254
result = client.describe_query("SELECT User { name, email }")
255
256
print(f"Output cardinality: {result.output_cardinality}")
257
print(f"Output type: {result.output_type.name}")
258
print(f"Capabilities required: {result.capabilities}")
259
260
# Describe parameterized query
261
result = client.describe_query(
262
"SELECT User { name } FILTER .id = <uuid>$user_id",
263
user_id="123e4567-e89b-12d3-a456-426614174000"
264
)
265
266
if result.input_type:
267
print(f"Input type: {result.input_type.name}")
268
```
269
270
### Complex Type Analysis
271
272
```python
273
import edgedb
274
275
client = edgedb.create_client()
276
277
# Describe complex nested query
278
query = """
279
SELECT Article {
280
title,
281
content,
282
author: { name, email },
283
tags,
284
comments: {
285
content,
286
author: { name },
287
replies: { content, author: { name } }
288
}
289
}
290
FILTER .published = true
291
"""
292
293
result = client.describe_query(query)
294
295
def analyze_type(type_desc, level=0):
296
"""Recursively analyze type structure."""
297
indent = " " * level
298
print(f"{indent}Type: {type_desc.name or 'anonymous'}")
299
300
if hasattr(type_desc, 'element_type'):
301
print(f"{indent} Element type:")
302
analyze_type(type_desc.element_type, level + 2)
303
304
if hasattr(type_desc, 'fields'):
305
print(f"{indent} Fields:")
306
for field_name, field_desc in type_desc.fields.items():
307
print(f"{indent} {field_name}:")
308
analyze_type(field_desc.type, level + 3)
309
310
if result.output_type:
311
analyze_type(result.output_type)
312
```
313
314
### Schema Version Tracking
315
316
```python
317
import edgedb
318
319
client = edgedb.create_client()
320
321
# Get current schema version
322
version = client.get_schema_version()
323
print(f"Current schema version: {version}")
324
325
# Store version for comparison
326
stored_version = version
327
328
# ... perform schema migrations ...
329
330
# Check if schema changed
331
new_version = client.get_schema_version()
332
if new_version != stored_version:
333
print("Schema has been updated")
334
print(f"Old version: {stored_version}")
335
print(f"New version: {new_version}")
336
```
337
338
### Query Capability Analysis
339
340
```python
341
import edgedb
342
343
client = edgedb.create_client()
344
345
def analyze_query_capabilities(query: str):
346
"""Analyze what capabilities a query requires."""
347
result = client.describe_query(query)
348
349
capabilities = result.capabilities
350
required_caps = []
351
352
# Check specific capability flags
353
if capabilities & edgedb.Capability.MODIFICATIONS:
354
required_caps.append("MODIFICATIONS")
355
if capabilities & edgedb.Capability.SESSION_CONFIG:
356
required_caps.append("SESSION_CONFIG")
357
if capabilities & edgedb.Capability.TRANSACTION:
358
required_caps.append("TRANSACTION")
359
if capabilities & edgedb.Capability.DDL:
360
required_caps.append("DDL")
361
362
print(f"Query: {query}")
363
print(f"Required capabilities: {', '.join(required_caps) or 'None'}")
364
return required_caps
365
366
# Analyze different types of queries
367
analyze_query_capabilities("SELECT User { name }")
368
analyze_query_capabilities("INSERT User { name := 'Alice' }")
369
analyze_query_capabilities("CREATE TYPE NewType { name: str }")
370
analyze_query_capabilities("CONFIGURE SESSION SET query_execution_timeout := '30s'")
371
```
372
373
### Type System Exploration
374
375
```python
376
import edgedb
377
378
client = edgedb.create_client()
379
380
def explore_object_type(type_name: str):
381
"""Explore the structure of an object type."""
382
383
query = f"""
384
SELECT schema::ObjectType {{
385
name,
386
properties: {{
387
name,
388
target: {{ name }},
389
cardinality,
390
required
391
}},
392
links: {{
393
name,
394
target: {{ name }},
395
cardinality,
396
required
397
}}
398
}}
399
FILTER .name = '{type_name}'
400
"""
401
402
type_info = client.query_single(query)
403
404
if not type_info:
405
print(f"Type '{type_name}' not found")
406
return
407
408
print(f"Object Type: {type_info.name}")
409
410
print("\nProperties:")
411
for prop in type_info.properties:
412
cardinality = "REQUIRED" if prop.required else "OPTIONAL"
413
print(f" {prop.name}: {prop.target.name} [{cardinality}, {prop.cardinality}]")
414
415
print("\nLinks:")
416
for link in type_info.links:
417
cardinality = "REQUIRED" if link.required else "OPTIONAL"
418
print(f" {link.name} -> {link.target.name} [{cardinality}, {link.cardinality}]")
419
420
# Explore specific types
421
explore_object_type("User")
422
explore_object_type("Article")
423
```
424
425
### Schema Validation
426
427
```python
428
import edgedb
429
430
def validate_schema_compatibility(client, expected_types):
431
"""Validate that schema contains expected types and structure."""
432
433
issues = []
434
435
for type_name in expected_types:
436
try:
437
# Check if type exists
438
query = f"SELECT schema::ObjectType FILTER .name = '{type_name}'"
439
type_exists = client.query_single(query)
440
441
if not type_exists:
442
issues.append(f"Missing type: {type_name}")
443
continue
444
445
# Validate type structure
446
result = client.describe_query(f"SELECT {type_name} {{ * }}")
447
448
if not result.output_type:
449
issues.append(f"Cannot describe type: {type_name}")
450
451
except edgedb.EdgeDBError as e:
452
issues.append(f"Error checking type {type_name}: {e}")
453
454
return issues
455
456
# Usage
457
client = edgedb.create_client()
458
expected_types = ["User", "Article", "Comment", "Tag"]
459
schema_issues = validate_schema_compatibility(client, expected_types)
460
461
if schema_issues:
462
print("Schema validation issues:")
463
for issue in schema_issues:
464
print(f" - {issue}")
465
else:
466
print("Schema validation passed")
467
```
468
469
### Query Optimization Analysis
470
471
```python
472
import edgedb
473
from typing import Dict, Any
474
475
def analyze_query_performance(client, queries: Dict[str, str]):
476
"""Analyze queries for performance characteristics."""
477
478
analysis_results = {}
479
480
for name, query in queries.items():
481
try:
482
result = client.describe_query(query)
483
484
analysis = {
485
'cardinality': result.output_cardinality.value,
486
'capabilities': result.capabilities,
487
'has_filters': 'FILTER' in query.upper(),
488
'has_order': 'ORDER BY' in query.upper(),
489
'has_limit': 'LIMIT' in query.upper(),
490
'complexity_score': calculate_complexity(query)
491
}
492
493
analysis_results[name] = analysis
494
495
except edgedb.EdgeDBError as e:
496
analysis_results[name] = {'error': str(e)}
497
498
return analysis_results
499
500
def calculate_complexity(query: str) -> int:
501
"""Simple query complexity score."""
502
score = 0
503
score += query.count('SELECT') * 1
504
score += query.count('FILTER') * 2
505
score += query.count('ORDER BY') * 1
506
score += query.count('GROUP BY') * 3
507
score += query.count('{') * 1 # Nested selections
508
return score
509
510
# Usage
511
queries = {
512
'simple_user_query': "SELECT User { name, email }",
513
'complex_article_query': """
514
SELECT Article {
515
title,
516
author: { name },
517
comments: { content, author: { name } },
518
tag_count := count(.tags)
519
}
520
FILTER .published = true
521
ORDER BY .created_at DESC
522
LIMIT 10
523
""",
524
'aggregation_query': """
525
SELECT User {
526
name,
527
article_count := count(.articles),
528
avg_comment_count := math::mean(.articles.comment_count)
529
}
530
GROUP BY .department
531
"""
532
}
533
534
client = edgedb.create_client()
535
analysis = analyze_query_performance(client, queries)
536
537
for query_name, results in analysis.items():
538
print(f"\n{query_name}:")
539
if 'error' in results:
540
print(f" Error: {results['error']}")
541
else:
542
print(f" Cardinality: {results['cardinality']}")
543
print(f" Complexity Score: {results['complexity_score']}")
544
print(f" Has Filters: {results['has_filters']}")
545
print(f" Has Ordering: {results['has_order']}")
546
```
547
548
### Dynamic Query Building with Introspection
549
550
```python
551
import edgedb
552
553
def build_filtered_query(client, type_name, filters=None, fields=None):
554
"""Build query dynamically based on type introspection."""
555
556
# Get type information
557
type_query = f"""
558
SELECT schema::ObjectType {{
559
properties: {{ name, target: {{ name }} }},
560
links: {{ name, target: {{ name }} }}
561
}}
562
FILTER .name = '{type_name}'
563
"""
564
565
type_info = client.query_single(type_query)
566
if not type_info:
567
raise ValueError(f"Type {type_name} not found")
568
569
# Build field selection
570
if fields is None:
571
# Select all scalar properties by default
572
fields = [prop.name for prop in type_info.properties
573
if prop.target.name in ['std::str', 'std::int64', 'std::bool', 'std::datetime']]
574
575
field_selection = "{ " + ", ".join(fields) + " }"
576
577
# Build base query
578
query = f"SELECT {type_name} {field_selection}"
579
580
# Add filters if provided
581
if filters:
582
filter_parts = []
583
for field, value in filters.items():
584
if isinstance(value, str):
585
filter_parts.append(f".{field} = '{value}'")
586
else:
587
filter_parts.append(f".{field} = {value}")
588
589
if filter_parts:
590
query += " FILTER " + " AND ".join(filter_parts)
591
592
return query
593
594
# Usage
595
client = edgedb.create_client()
596
597
# Build query dynamically
598
query = build_filtered_query(
599
client,
600
"User",
601
filters={"active": True, "role": "admin"},
602
fields=["name", "email", "created_at"]
603
)
604
605
print(f"Generated query: {query}")
606
results = client.query(query)
607
```