Coverage for /pythoncovmergedfiles/medio/medio/src/pydantic/pydantic/json_schema.py: 18%

1from __future__ import annotations as _annotations 

2 

3import inspect 

4import math 

5import re 

6import sys 

7import warnings 

8from dataclasses import is_dataclass 

9from enum import Enum 

10from typing import ( 

11 TYPE_CHECKING, 

12 Any, 

13 Callable, 

14 Counter, 

15 Dict, 

16 Iterable, 

17 List, 

18 NewType, 

19 Sequence, 

20 Tuple, 

21 Type, 

22 Union, 

23 cast, 

24) 

25from weakref import WeakKeyDictionary 

26 

27import pydantic_core 

28from typing_extensions import Literal 

29 

30from pydantic._internal._json_schema_shared import GenerateJsonSchemaHandler 

31 

32from ._internal import _core_metadata, _core_utils, _json_schema_shared, _typing_extra 

33from .errors import PydanticInvalidForJsonSchema, PydanticUserError 

34 

35if TYPE_CHECKING: 

36 from pydantic_core import CoreSchema, CoreSchemaType, core_schema 

37 

38 from . import ConfigDict 

39 from ._internal._dataclasses import PydanticDataclass 

40 from .main import BaseModel 

41 

42 

43JsonSchemaValue = _json_schema_shared.JsonSchemaValue 

44# re export GetJsonSchemaHandler 

45GetJsonSchemaHandler = _json_schema_shared.GetJsonSchemaHandler 

46 

47 

48def update_json_schema(schema: JsonSchemaValue, updates: dict[str, Any]) -> JsonSchemaValue: 

49 """ 

50 A convenience function useful for creating `js_modify_function` functions that just set values for some keys. 

51 

52 TODO: This is basically just a wrapper for dict.update that returns the dict. 

53 Would it be better to just make this a less-"domain-specific" utility function? 

54 """ 

55 schema.update(updates) 

56 return schema 

57 

58 

59# These are "kind" labels that can be used to control warnings. See `GenerateJsonSchema.render_warning_message` 

60JsonSchemaWarningKind = Literal['skipped-choice', 'non-serializable-default'] 

61 

62 

63class PydanticJsonSchemaWarning(UserWarning): 

64 """ 

65 This class is used to emit warnings produced during JSON schema generation. 

66 See the `GenerateJsonSchema.emit_warning` and `GenerateJsonSchema.render_warning_message` 

67 methods for more details; these can be overridden to control warning behavior 

68 """ 

69 

70 

71# ##### JSON Schema Generation ##### 

72DEFAULT_REF_TEMPLATE = '#/$defs/{model}' 

73 

74# There are three types of references relevant to building JSON schemas: 

75# 1. core_schema "ref" values; these are not exposed as part of the JSON schema 

76# * these might look like the fully qualified path of a model, its id, or something similar 

77CoreRef = NewType('CoreRef', str) 

78# 2. keys of the "definitions" object that will eventually go into the JSON schema 

79# * by default, these look like "MyModel", though may change in the presence of collisions 

80# * eventually, we may want to make it easier to modify the way these names are generated 

81DefsRef = NewType('DefsRef', str) 

82# 3. the values corresponding to the "$ref" key in the schema 

83# * By default, these look like "#/$defs/MyModel", as in {"$ref": "#/$defs/MyModel"} 

84JsonRef = NewType('JsonRef', str) 

85 

86 

87class GenerateJsonSchema: 

88 # See https://json-schema.org/understanding-json-schema/reference/schema.html#id4 for more info about dialects 

89 schema_dialect = 'https://json-schema.org/draft/2020-12/schema' 

90 

91 # `self.render_warning_message` will do nothing if its argument `kind` is in `ignored_warning_kinds`; 

92 # this value can be modified on subclasses to easily control which warnings are emitted 

93 ignored_warning_kinds: set[JsonSchemaWarningKind] = {'skipped-choice'} 

94 

95 def __init__(self, by_alias: bool = True, ref_template: str = DEFAULT_REF_TEMPLATE): 

96 self.by_alias = by_alias 

97 self.ref_template = ref_template 

98 

99 self.core_to_json_refs: dict[CoreRef, JsonRef] = {} 

100 self.core_to_defs_refs: dict[CoreRef, DefsRef] = {} 

101 self.defs_to_core_refs: dict[DefsRef, CoreRef] = {} 

102 self.json_to_defs_refs: dict[JsonRef, DefsRef] = {} 

103 

104 self.definitions: dict[DefsRef, JsonSchemaValue] = {} 

105 

106 # When collisions are detected, we choose a non-colliding name 

107 # during generation, but we also track the colliding tag so that it 

108 # can be remapped for the first occurrence at the end of the process 

109 self.collisions: set[DefsRef] = set() 

110 self.defs_ref_fallbacks: dict[CoreRef, list[DefsRef]] = {} 

111 

112 self._schema_type_to_method = self.build_schema_type_to_method() 

113 

114 # This changes to True after generating a schema, to prevent issues caused by accidental re-use 

115 # of a single instance of a schema generator 

116 self._used = False 

117 

118 def build_schema_type_to_method(self) -> dict[CoreSchemaType, Callable[[CoreSchema], JsonSchemaValue]]: 

119 mapping: dict[CoreSchemaType, Callable[[CoreSchema], JsonSchemaValue]] = {} 

120 for key in _typing_extra.all_literal_values(pydantic_core.CoreSchemaType): # type: ignore[arg-type] 

121 method_name = f"{key.replace('-', '_')}_schema" 

122 try: 

123 mapping[key] = getattr(self, method_name) 

124 except AttributeError as e: 

125 raise TypeError( 

126 f'No method for generating JsonSchema for core_schema.type={key!r} ' 

127 f'(expected: {type(self).__name__}.{method_name})' 

128 ) from e 

129 return mapping 

130 

131 def generate_definitions(self, schemas: list[CoreSchema]) -> dict[DefsRef, JsonSchemaValue]: 

132 """ 

133 Given a list of core_schema, generate all JSON schema definitions, and return the generated definitions. 

134 """ 

135 if self._used: 

136 raise PydanticUserError( 

137 'This JSON schema generator has already been used to generate a JSON schema. ' 

138 f'You must create a new instance of {type(self).__name__} to generate a new JSON schema.', 

139 code='json-schema-already-used', 

140 ) 

141 for schema in schemas: 

142 self.generate_inner(schema) 

143 

144 self.resolve_collisions({}) 

145 

146 self._used = True 

147 return self.definitions 

148 

149 def generate(self, schema: CoreSchema) -> JsonSchemaValue: 

150 if self._used: 

151 raise PydanticUserError( 

152 'This JSON schema generator has already been used to generate a JSON schema. ' 

153 f'You must create a new instance of {type(self).__name__} to generate a new JSON schema.', 

154 code='json-schema-already-used', 

155 ) 

156 

157 json_schema = self.generate_inner(schema) 

158 json_ref_counts = self.get_json_ref_counts(json_schema) 

159 

160 # Remove the top-level $ref if present; note that the _generate method already ensures there are no sibling keys 

161 ref = json_schema.get('$ref') 

162 while ref is not None: # may need to unpack multiple levels 

163 ref = JsonRef(ref) 

164 ref_json_schema = self.get_schema_from_definitions(ref) 

165 if json_ref_counts[ref] > 1 or ref_json_schema is None: 

166 # Keep the ref, but use an allOf to remove the top level $ref 

167 json_schema = {'allOf': [{'$ref': ref}]} 

168 else: 

169 # "Unpack" the ref since this is the only reference 

170 json_schema = ref_json_schema.copy() # copy to prevent recursive dict reference 

171 json_ref_counts[ref] -= 1 

172 ref = json_schema.get('$ref') 

173 

174 # Remove any definitions that, thanks to $ref-substitution, are no longer present. 

175 # I think this should only _possibly_ apply to the root model, though I'm not 100% sure. 

176 # It might be safe to remove this logic, but I'm keeping it for now 

177 all_json_refs = list(self.json_to_defs_refs.keys()) 

178 for k in all_json_refs: 

179 if json_ref_counts[k] < 1: 

180 del self.definitions[self.json_to_defs_refs[k]] 

181 

182 json_schema = self.resolve_collisions(json_schema) 

183 if self.definitions: 

184 json_schema['$defs'] = self.definitions 

185 

186 # For now, we will not set the $schema key. However, if desired, this can be easily added by overriding 

187 # this method and adding the following line after a call to super().generate(schema): 

188 # json_schema['$schema'] = self.schema_dialect 

189 

190 self._used = True 

191 return json_schema 

192 

193 def generate_inner(self, schema: _core_metadata.CoreSchemaOrField) -> JsonSchemaValue: 

194 # If a schema with the same CoreRef has been handled, just return a reference to it 

195 # Note that this assumes that it will _never_ be the case that the same CoreRef is used 

196 # on types that should have different JSON schemas 

197 if 'ref' in schema: 

198 core_ref = CoreRef(schema['ref']) # type: ignore[typeddict-item] 

199 if core_ref in self.core_to_json_refs: 

200 return {'$ref': self.core_to_json_refs[core_ref]} 

201 

202 # Generate the JSON schema, accounting for the json_schema_override and core_schema_override 

203 metadata_handler = _core_metadata.CoreMetadataHandler(schema) 

204 

205 def handler_func(schema_or_field: _core_metadata.CoreSchemaOrField) -> JsonSchemaValue: 

206 # Generate the core-schema-type-specific bits of the schema generation: 

207 if _core_utils.is_typed_dict_field(schema_or_field): 

208 json_schema = self.typed_dict_field_schema(schema_or_field) 

209 elif _core_utils.is_dataclass_field(schema_or_field): 

210 json_schema = self.dataclass_field_schema(schema_or_field) 

211 elif _core_utils.is_core_schema(schema_or_field): # Ideally we wouldn't need this redundant typeguard.. 

212 generate_for_schema_type = self._schema_type_to_method[schema_or_field['type']] 

213 json_schema = generate_for_schema_type(schema_or_field) 

214 else: 

215 raise TypeError(f'Unexpected schema type: schema={schema_or_field}') 

216 # Populate the definitions 

217 if 'ref' in schema: 

218 core_ref = CoreRef(schema['ref']) # type: ignore[typeddict-item] 

219 defs_ref, ref_json_schema = self.get_cache_defs_ref_schema(core_ref) 

220 self.definitions[defs_ref] = json_schema 

221 json_schema = ref_json_schema 

222 return json_schema 

223 

224 current_handler = GenerateJsonSchemaHandler(self, handler_func) 

225 

226 for js_modify_function in metadata_handler.metadata.get('pydantic_js_functions', ()): 

227 

228 def new_handler_func( 

229 schema_or_field: _core_metadata.CoreSchemaOrField, 

230 current_handler: _core_metadata.GetJsonSchemaHandler = current_handler, 

231 js_modify_function: _core_metadata.GetJsonSchemaFunction = js_modify_function, 

232 ) -> JsonSchemaValue: 

233 return js_modify_function(schema_or_field, current_handler) 

234 

235 current_handler = GenerateJsonSchemaHandler(self, new_handler_func) 

236 

237 return current_handler(schema) 

238 

239 # ### Schema generation methods 

240 def any_schema(self, schema: core_schema.AnySchema) -> JsonSchemaValue: 

241 return {} 

242 

243 def none_schema(self, schema: core_schema.NoneSchema) -> JsonSchemaValue: 

244 return {'type': 'null'} 

245 

246 def bool_schema(self, schema: core_schema.BoolSchema) -> JsonSchemaValue: 

247 return {'type': 'boolean'} 

248 

249 def int_schema(self, schema: core_schema.IntSchema) -> JsonSchemaValue: 

250 json_schema: dict[str, Any] = {'type': 'integer'} 

251 self.update_with_validations(json_schema, schema, self.ValidationsMapping.numeric) 

252 json_schema = {k: v for k, v in json_schema.items() if v not in {math.inf, -math.inf}} 

253 return json_schema 

254 

255 def float_schema(self, schema: core_schema.FloatSchema) -> JsonSchemaValue: 

256 json_schema: dict[str, Any] = {'type': 'number'} 

257 self.update_with_validations(json_schema, schema, self.ValidationsMapping.numeric) 

258 json_schema = {k: v for k, v in json_schema.items() if v not in {math.inf, -math.inf}} 

259 return json_schema 

260 

261 def str_schema(self, schema: core_schema.StringSchema) -> JsonSchemaValue: 

262 json_schema = {'type': 'string'} 

263 self.update_with_validations(json_schema, schema, self.ValidationsMapping.string) 

264 return json_schema 

265 

266 def bytes_schema(self, schema: core_schema.BytesSchema) -> JsonSchemaValue: 

267 json_schema = {'type': 'string', 'format': 'binary'} 

268 self.update_with_validations(json_schema, schema, self.ValidationsMapping.bytes) 

269 return json_schema 

270 

271 def date_schema(self, schema: core_schema.DateSchema) -> JsonSchemaValue: 

272 json_schema = {'type': 'string', 'format': 'date'} 

273 self.update_with_validations(json_schema, schema, self.ValidationsMapping.date) 

274 return json_schema 

275 

276 def time_schema(self, schema: core_schema.TimeSchema) -> JsonSchemaValue: 

277 return {'type': 'string', 'format': 'time'} 

278 

279 def datetime_schema(self, schema: core_schema.DatetimeSchema) -> JsonSchemaValue: 

280 return {'type': 'string', 'format': 'date-time'} 

281 

282 def timedelta_schema(self, schema: core_schema.TimedeltaSchema) -> JsonSchemaValue: 

283 # It's weird that this schema has 'type': 'number' but also specifies a 'format'. 

284 # Relevant issue: https://github.com/pydantic/pydantic/issues/5034 

285 # TODO: Probably should just change this to str (look at readme intro for speeddate) 

286 return {'type': 'number', 'format': 'time-delta'} 

287 

288 def literal_schema(self, schema: core_schema.LiteralSchema) -> JsonSchemaValue: 

289 expected = [v.value if isinstance(v, Enum) else v for v in schema['expected']] 

290 

291 if len(expected) == 1: 

292 return {'const': expected[0]} 

293 else: 

294 return {'enum': expected} 

295 

296 def is_instance_schema(self, schema: core_schema.IsInstanceSchema) -> JsonSchemaValue: 

297 return self.handle_invalid_for_json_schema(schema, f'core_schema.IsInstanceSchema ({schema["cls"]})') 

298 

299 def is_subclass_schema(self, schema: core_schema.IsSubclassSchema) -> JsonSchemaValue: 

300 return {} # TODO: This was for compatibility with V1 -- is this the right thing to do? 

301 

302 def callable_schema(self, schema: core_schema.CallableSchema) -> JsonSchemaValue: 

303 return self.handle_invalid_for_json_schema(schema, 'core_schema.CallableSchema') 

304 

305 def list_schema(self, schema: core_schema.ListSchema) -> JsonSchemaValue: 

306 items_schema = {} if 'items_schema' not in schema else self.generate_inner(schema['items_schema']) 

307 json_schema = {'type': 'array', 'items': items_schema} 

308 self.update_with_validations(json_schema, schema, self.ValidationsMapping.array) 

309 return json_schema 

310 

311 def tuple_positional_schema(self, schema: core_schema.TuplePositionalSchema) -> JsonSchemaValue: 

312 json_schema: JsonSchemaValue = {'type': 'array'} 

313 json_schema['minItems'] = len(schema['items_schema']) 

314 prefixItems = [self.generate_inner(item) for item in schema['items_schema']] 

315 if prefixItems: 

316 json_schema['prefixItems'] = prefixItems 

317 if 'extra_schema' in schema: 

318 json_schema['items'] = self.generate_inner(schema['extra_schema']) 

319 else: 

320 json_schema['maxItems'] = len(schema['items_schema']) 

321 self.update_with_validations(json_schema, schema, self.ValidationsMapping.array) 

322 return json_schema 

323 

324 def tuple_variable_schema(self, schema: core_schema.TupleVariableSchema) -> JsonSchemaValue: 

325 json_schema: JsonSchemaValue = {'type': 'array', 'items': {}} 

326 if 'items_schema' in schema: 

327 json_schema['items'] = self.generate_inner(schema['items_schema']) 

328 self.update_with_validations(json_schema, schema, self.ValidationsMapping.array) 

329 return json_schema 

330 

331 def set_schema(self, schema: core_schema.SetSchema) -> JsonSchemaValue: 

332 return self._common_set_schema(schema) 

333 

334 def frozenset_schema(self, schema: core_schema.FrozenSetSchema) -> JsonSchemaValue: 

335 return self._common_set_schema(schema) 

336 

337 def _common_set_schema(self, schema: core_schema.SetSchema | core_schema.FrozenSetSchema) -> JsonSchemaValue: 

338 items_schema = {} if 'items_schema' not in schema else self.generate_inner(schema['items_schema']) 

339 json_schema = {'type': 'array', 'uniqueItems': True, 'items': items_schema} 

340 self.update_with_validations(json_schema, schema, self.ValidationsMapping.array) 

341 return json_schema 

342 

343 def generator_schema(self, schema: core_schema.GeneratorSchema) -> JsonSchemaValue: 

344 items_schema = {} if 'items_schema' not in schema else self.generate_inner(schema['items_schema']) 

345 json_schema = {'type': 'array', 'items': items_schema} 

346 self.update_with_validations(json_schema, schema, self.ValidationsMapping.array) 

347 return json_schema 

348 

349 def dict_schema(self, schema: core_schema.DictSchema) -> JsonSchemaValue: 

350 json_schema: JsonSchemaValue = {'type': 'object'} 

351 

352 keys_schema = self.generate_inner(schema['keys_schema']).copy() if 'keys_schema' in schema else {} 

353 keys_pattern = keys_schema.pop('pattern', None) 

354 

355 values_schema = self.generate_inner(schema['values_schema']).copy() if 'values_schema' in schema else {} 

356 values_schema.pop('title', None) # don't give a title to the additionalProperties 

357 if values_schema or keys_pattern is not None: # don't add additionalProperties if it's empty 

358 if keys_pattern is None: 

359 json_schema['additionalProperties'] = values_schema 

360 else: 

361 json_schema['patternProperties'] = {keys_pattern: values_schema} 

362 

363 self.update_with_validations(json_schema, schema, self.ValidationsMapping.object) 

364 return json_schema 

365 

366 def _function_schema( 

367 self, 

368 schema: _core_utils.AnyFunctionSchema, 

369 ) -> JsonSchemaValue: 

370 if _core_utils.is_function_with_inner_schema(schema): 

371 # I'm not sure if this might need to be different if the function's mode is 'before' 

372 return self.generate_inner(schema['schema']) 

373 # function-plain 

374 return self.handle_invalid_for_json_schema( 

375 schema, f'core_schema.PlainValidatorFunctionSchema ({schema["function"]})' 

376 ) 

377 

378 def function_before_schema(self, schema: core_schema.BeforeValidatorFunctionSchema) -> JsonSchemaValue: 

379 return self._function_schema(schema) 

380 

381 def function_after_schema(self, schema: core_schema.AfterValidatorFunctionSchema) -> JsonSchemaValue: 

382 return self._function_schema(schema) 

383 

384 def function_plain_schema(self, schema: core_schema.PlainValidatorFunctionSchema) -> JsonSchemaValue: 

385 return self._function_schema(schema) 

386 

387 def function_wrap_schema(self, schema: core_schema.WrapValidatorFunctionSchema) -> JsonSchemaValue: 

388 return self._function_schema(schema) 

389 

390 def default_schema(self, schema: core_schema.WithDefaultSchema) -> JsonSchemaValue: 

391 json_schema = self.generate_inner(schema['schema']) 

392 

393 if 'default' in schema: 

394 default = schema['default'] 

395 elif 'default_factory' in schema: 

396 default = schema['default_factory']() 

397 else: 

398 raise ValueError('`schema` has neither default nor default_factory') 

399 

400 try: 

401 encoded_default = self.encode_default(default) 

402 except pydantic_core.PydanticSerializationError: 

403 self.emit_warning( 

404 'non-serializable-default', 

405 f'Default value {default} is not JSON serializable; excluding default from JSON schema', 

406 ) 

407 # Return the inner schema, as though there was no default 

408 return json_schema 

409 

410 if '$ref' in json_schema: 

411 # Since reference schemas do not support child keys, we wrap the reference schema in a single-case allOf: 

412 return {'allOf': [json_schema], 'default': encoded_default} 

413 else: 

414 json_schema['default'] = encoded_default 

415 return json_schema 

416 

417 def nullable_schema(self, schema: core_schema.NullableSchema) -> JsonSchemaValue: 

418 null_schema = {'type': 'null'} 

419 inner_json_schema = self.generate_inner(schema['schema']) 

420 

421 if inner_json_schema == null_schema: 

422 return null_schema 

423 else: 

424 # Thanks to the equality check against `null_schema` above, I think 'oneOf' would also be valid here; 

425 # I'll use 'anyOf' for now, but it could be changed it if it would work better with some external tooling 

426 return self.get_flattened_anyof([inner_json_schema, null_schema]) 

427 

428 def union_schema(self, schema: core_schema.UnionSchema) -> JsonSchemaValue: 

429 generated: list[JsonSchemaValue] = [] 

430 

431 choices = schema['choices'] 

432 for s in choices: 

433 try: 

434 generated.append(self.generate_inner(s)) 

435 except PydanticInvalidForJsonSchema as exc: 

436 self.emit_warning('skipped-choice', exc.message) 

437 if len(generated) == 1: 

438 return generated[0] 

439 return self.get_flattened_anyof(generated) 

440 

441 def tagged_union_schema(self, schema: core_schema.TaggedUnionSchema) -> JsonSchemaValue: 

442 generated: dict[str, JsonSchemaValue] = {} 

443 for k, v in schema['choices'].items(): 

444 if not isinstance(v, (str, int)): 

445 try: 

446 # Use str(k) since keys must be strings for json; while not technically correct, 

447 # it's the closest that can be represented in valid JSON 

448 generated[str(k)] = self.generate_inner(v).copy() 

449 except PydanticInvalidForJsonSchema as exc: 

450 self.emit_warning('skipped-choice', exc.message) 

451 

452 # Populate the schema with any "indirect" references 

453 for k, v in schema['choices'].items(): 

454 if isinstance(v, (str, int)): 

455 while isinstance(schema['choices'][v], (str, int)): 

456 v = schema['choices'][v] 

457 assert isinstance(v, (int, str)) 

458 if str(v) in generated: 

459 # while it might seem unnecessary to check `if str(v) in generated`, a PydanticInvalidForJsonSchema 

460 # may have been raised above, which would mean that the schema we want to reference won't be present 

461 generated[str(k)] = generated[str(v)] 

462 

463 one_of_choices = _deduplicate_schemas(generated.values()) 

464 json_schema: JsonSchemaValue = {'oneOf': one_of_choices} 

465 

466 # This reflects the v1 behavior; TODO: we should make it possible to exclude OpenAPI stuff from the JSON schema 

467 openapi_discriminator = self._extract_discriminator(schema, one_of_choices) 

468 if openapi_discriminator is not None: 

469 json_schema['discriminator'] = { 

470 'propertyName': openapi_discriminator, 

471 'mapping': {k: v.get('$ref', v) for k, v in generated.items()}, 

472 } 

473 

474 return json_schema 

475 

476 def _extract_discriminator( 

477 self, schema: core_schema.TaggedUnionSchema, one_of_choices: list[_JsonDict] 

478 ) -> str | None: 

479 """ 

480 Extract a compatible OpenAPI discriminator from the schema and one_of choices that end up in the final schema. 

481 """ 

482 openapi_discriminator: str | None = None 

483 if 'discriminator' not in schema: 

484 return None 

485 

486 if isinstance(schema['discriminator'], str): 

487 return schema['discriminator'] 

488 

489 if isinstance(schema['discriminator'], list): 

490 # If the discriminator is a single item list containing a string, that is equivalent to the string case 

491 if len(schema['discriminator']) == 1 and isinstance(schema['discriminator'][0], str): 

492 return schema['discriminator'][0] 

493 # When an alias is used that is different from the field name, the discriminator will be a list of single 

494 # str lists, one for the attribute and one for the actual alias. The logic here will work even if there is 

495 # more than one possible attribute, and looks for whether a single alias choice is present as a documented 

496 # property on all choices. If so, that property will be used as the OpenAPI discriminator. 

497 for alias_path in schema['discriminator']: 

498 if not isinstance(alias_path, list): 

499 break # this means that the discriminator is not a list of alias paths 

500 if len(alias_path) != 1: 

501 continue # this means that the "alias" does not represent a single field 

502 alias = alias_path[0] 

503 if not isinstance(alias, str): 

504 continue # this means that the "alias" does not represent a field 

505 alias_is_present_on_all_choices = True 

506 for choice in one_of_choices: 

507 while '$ref' in choice: 

508 assert isinstance(choice['$ref'], str) 

509 choice = self.get_schema_from_definitions(JsonRef(choice['$ref'])) or {} 

510 properties = choice.get('properties', {}) 

511 if not isinstance(properties, dict) or alias not in properties: 

512 alias_is_present_on_all_choices = False 

513 break 

514 if alias_is_present_on_all_choices: 

515 openapi_discriminator = alias 

516 break 

517 return openapi_discriminator 

518 

519 def chain_schema(self, schema: core_schema.ChainSchema) -> JsonSchemaValue: 

520 try: 

521 # Note: If we wanted to generate a schema for the _serialization_, would want to use the _last_ step: 

522 return self.generate_inner(schema['steps'][0]) 

523 except IndexError as e: 

524 raise ValueError('Cannot generate a JsonSchema for a zero-step ChainSchema') from e 

525 

526 def lax_or_strict_schema(self, schema: core_schema.LaxOrStrictSchema) -> JsonSchemaValue: 

527 """ 

528 LaxOrStrict will use the strict branch for serialization internally, 

529 unless it was overridden here. 

530 """ 

531 # TODO: Need to read the default value off of model config or whatever 

532 use_strict = schema.get('strict', False) # TODO: replace this default False 

533 # If your JSON schema fails to generate it is probably 

534 # because one of the following two branches failed. 

535 if use_strict: 

536 return self.generate_inner(schema['strict_schema']) 

537 else: 

538 return self.generate_inner(schema['lax_schema']) 

539 

540 def typed_dict_schema(self, schema: core_schema.TypedDictSchema) -> JsonSchemaValue: 

541 named_required_fields = [ 

542 (k, v['required'], v) for k, v in schema['fields'].items() # type: ignore # required is always populated 

543 ] 

544 return self._named_required_fields_schema(named_required_fields) 

545 

546 def _named_required_fields_schema( 

547 self, named_required_fields: Sequence[tuple[str, bool, core_schema.TypedDictField | core_schema.DataclassField]] 

548 ) -> JsonSchemaValue: 

549 properties: dict[str, JsonSchemaValue] = {} 

550 required_fields: list[str] = [] 

551 for name, required, field in named_required_fields: 

552 if self.by_alias: 

553 alias: Any = field.get('validation_alias', name) 

554 if isinstance(alias, str): 

555 name = alias 

556 elif isinstance(alias, list): 

557 alias = cast('list[str] | str', alias) 

558 for path in alias: 

559 if isinstance(path, list) and len(path) == 1 and isinstance(path[0], str): 

560 # Use the first valid single-item string path; the code that constructs the alias array 

561 # should ensure the first such item is what belongs in the JSON schema 

562 name = path[0] 

563 break 

564 field_json_schema = self.generate_inner(field).copy() 

565 if 'title' not in field_json_schema and self.field_title_should_be_set(field): 

566 title = self.get_title_from_name(name) 

567 field_json_schema['title'] = title 

568 field_json_schema = self.handle_ref_overrides(field_json_schema) 

569 properties[name] = field_json_schema 

570 if required: 

571 required_fields.append(name) 

572 

573 json_schema = {'type': 'object', 'properties': properties} 

574 if required_fields: 

575 json_schema['required'] = required_fields # type: ignore 

576 return json_schema 

577 

578 def typed_dict_field_schema(self, schema: core_schema.TypedDictField) -> JsonSchemaValue: 

579 return self.generate_inner(schema['schema']) 

580 

581 def dataclass_field_schema(self, schema: core_schema.DataclassField) -> JsonSchemaValue: 

582 return self.generate_inner(schema['schema']) 

583 

584 def model_schema(self, schema: core_schema.ModelSchema) -> JsonSchemaValue: 

585 # We do not use schema['model'].model_json_schema() because it could lead to inconsistent refs handling, etc. 

586 json_schema = self.generate_inner(schema['schema']) 

587 

588 if 'config' in schema: 

589 title = schema['config'].get('title') 

590 forbid_additional_properties = schema['config'].get('extra_fields_behavior') == 'forbid' 

591 json_schema = self._update_class_schema(json_schema, title, forbid_additional_properties) 

592 

593 return json_schema 

594 

595 def _update_class_schema( 

596 self, json_schema: JsonSchemaValue, title: str | None, forbid_additional_properties: bool 

597 ) -> JsonSchemaValue: 

598 if '$ref' in json_schema: 

599 schema_to_update = self.get_schema_from_definitions(JsonRef(json_schema['$ref'])) or json_schema 

600 else: 

601 schema_to_update = json_schema 

602 

603 if title is not None: 

604 # referenced_schema['title'] = title 

605 schema_to_update.setdefault('title', title) 

606 

607 if forbid_additional_properties: 

608 schema_to_update['additionalProperties'] = False 

609 

610 return json_schema 

611 

612 def dataclass_args_schema(self, schema: core_schema.DataclassArgsSchema) -> JsonSchemaValue: 

613 named_required_fields = [ 

614 (field['name'], field['schema']['type'] != 'default', field) for field in schema['fields'] 

615 ] 

616 return self._named_required_fields_schema(named_required_fields) 

617 

618 def dataclass_schema(self, schema: core_schema.DataclassSchema) -> JsonSchemaValue: 

619 # TODO: Better-share this logic with model_schema 

620 # I'd prefer to clean this up _after_ we rework the approach to customizing dataclass JSON schema though 

621 

622 json_schema = self.generate_inner(schema['schema']).copy() 

623 

624 cls = schema['cls'] 

625 config: ConfigDict = getattr(cls, '__pydantic_config__', cast('ConfigDict', {})) 

626 

627 title = config.get('title') or cls.__name__ 

628 forbid_additional_properties = config.get('extra') == 'forbid' 

629 json_schema = self._update_class_schema(json_schema, title, forbid_additional_properties) 

630 

631 # Dataclass-specific handling of description 

632 if is_dataclass(cls) and not hasattr(cls, '__pydantic_validator__'): 

633 # vanilla dataclass; don't use cls.__doc__ as it will contain the class signature by default 

634 description = None 

635 else: 

636 description = None if cls.__doc__ is None else inspect.cleandoc(cls.__doc__) 

637 if description: 

638 json_schema['description'] = description 

639 

640 return json_schema 

641 

642 def arguments_schema(self, schema: core_schema.ArgumentsSchema) -> JsonSchemaValue: 

643 metadata = _core_metadata.CoreMetadataHandler(schema).metadata 

644 prefer_positional = metadata.get('pydantic_js_prefer_positional_arguments') 

645 

646 arguments = schema['arguments_schema'] 

647 kw_only_arguments = [a for a in arguments if a.get('mode') == 'keyword_only'] 

648 kw_or_p_arguments = [a for a in arguments if a.get('mode') in {'positional_or_keyword', None}] 

649 p_only_arguments = [a for a in arguments if a.get('mode') == 'positional_only'] 

650 var_args_schema = schema.get('var_args_schema') 

651 var_kwargs_schema = schema.get('var_kwargs_schema') 

652 

653 if prefer_positional: 

654 positional_possible = not kw_only_arguments and not var_kwargs_schema 

655 if positional_possible: 

656 return self.p_arguments_schema(p_only_arguments + kw_or_p_arguments, var_args_schema) 

657 

658 keyword_possible = not p_only_arguments and not var_args_schema 

659 if keyword_possible: 

660 return self.kw_arguments_schema(kw_or_p_arguments + kw_only_arguments, var_kwargs_schema) 

661 

662 if not prefer_positional: 

663 positional_possible = not kw_only_arguments and not var_kwargs_schema 

664 if positional_possible: 

665 return self.p_arguments_schema(p_only_arguments + kw_or_p_arguments, var_args_schema) 

666 

667 raise PydanticInvalidForJsonSchema( 

668 'Unable to generate JSON schema for arguments validator with positional only and keyword only arguments' 

669 ) 

670 

671 def kw_arguments_schema( 

672 self, arguments: list[core_schema.ArgumentsParameter], var_kwargs_schema: CoreSchema | None 

673 ) -> JsonSchemaValue: 

674 properties: dict[str, JsonSchemaValue] = {} 

675 required: list[str] = [] 

676 for argument in arguments: 

677 name = self.get_argument_name(argument) 

678 argument_schema = self.generate_inner(argument['schema']).copy() 

679 argument_schema['title'] = self.get_title_from_name(name) 

680 properties[name] = argument_schema 

681 

682 if argument['schema']['type'] != 'default': 

683 # This assumes that if the argument has a default value, 

684 # the inner schema must be of type WithDefaultSchema. 

685 # I believe this is true, but I am not 100% sure 

686 required.append(name) 

687 

688 json_schema: JsonSchemaValue = {'type': 'object', 'properties': properties} 

689 if required: 

690 json_schema['required'] = required 

691 

692 if var_kwargs_schema: 

693 additional_properties_schema = self.generate_inner(var_kwargs_schema) 

694 if additional_properties_schema: 

695 json_schema['additionalProperties'] = additional_properties_schema 

696 else: 

697 json_schema['additionalProperties'] = False 

698 return json_schema 

699 

700 def p_arguments_schema( 

701 self, arguments: list[core_schema.ArgumentsParameter], var_args_schema: CoreSchema | None 

702 ) -> JsonSchemaValue: 

703 prefix_items: list[JsonSchemaValue] = [] 

704 min_items = 0 

705 

706 for argument in arguments: 

707 name = self.get_argument_name(argument) 

708 

709 argument_schema = self.generate_inner(argument['schema']).copy() 

710 argument_schema['title'] = self.get_title_from_name(name) 

711 prefix_items.append(argument_schema) 

712 

713 if argument['schema']['type'] != 'default': 

714 # This assumes that if the argument has a default value, 

715 # the inner schema must be of type WithDefaultSchema. 

716 # I believe this is true, but I am not 100% sure 

717 min_items += 1 

718 

719 json_schema: JsonSchemaValue = {'type': 'array', 'prefixItems': prefix_items} 

720 if min_items: 

721 json_schema['minItems'] = min_items 

722 

723 if var_args_schema: 

724 items_schema = self.generate_inner(var_args_schema) 

725 if items_schema: 

726 json_schema['items'] = items_schema 

727 else: 

728 json_schema['maxItems'] = len(prefix_items) 

729 

730 return json_schema 

731 

732 def get_argument_name(self, argument: core_schema.ArgumentsParameter) -> str: 

733 name = argument['name'] 

734 if self.by_alias: 

735 alias = argument.get('alias') 

736 if isinstance(alias, str): 

737 name = alias 

738 else: 

739 pass # might want to do something else? 

740 return name 

741 

742 def call_schema(self, schema: core_schema.CallSchema) -> JsonSchemaValue: 

743 return self.generate_inner(schema['arguments_schema']) 

744 

745 def custom_error_schema(self, schema: core_schema.CustomErrorSchema) -> JsonSchemaValue: 

746 return self.generate_inner(schema['schema']) 

747 

748 def json_schema(self, schema: core_schema.JsonSchema) -> JsonSchemaValue: 

749 # TODO: For v1 compatibility, we should probably be using `schema['schema']` to produce the schema.