schema ref_template (#1480)

* ignore Pipfile, .lock files

* add ref_template option

use a string.Template instead of a ref_prefix to allow for more varied`$ref`s to be created.
Template string is expected to have $model_name `identifier `

* formatting / linting

* add changes

* typo

* use string.format instead of string.Template

* remove ref_prefix default

if no `ref_prefix` provided, use the `template_default`

* use ref_template in field_singleton_schema

* fix test_schema_with_ref_template

* add parameters for `test_schema_with_refs`

test name change 
test for key error

* provide ref_template default argument

* fix linting

* fix mypy and coverage

* docs and correct model_schema usage

* fastAPI tests actualy caught an error

* linting

Co-authored-by: Samuel Colvin <s@muelcolvin.com>

.gitignore

@@ -4,6 +4,8 @@ venv/
 env36/
 env37/
 env38/
+Pipfile
+*.lock
 *.py[cod]
 *.egg-info/
 .python-version

changes/1479-kilo59.md

@@ -0,0 +1 @@
+Support `ref_template` when creating schema `$ref`s

docs/usage/schema.md

@@ -33,6 +33,10 @@ the `Field` class.
 The schema is generated by default using aliases as keys, but it can be generated using model
 property names instead by calling `MainModel.schema/schema_json(by_alias=False)`.
 
+The format of `$ref`s (`"#/definitions/FooBar"` above) can be altered by calling `schema()` or `schema_json()`
+with the `ref_template` keyword argument, e.g. `ApplePie.schema(ref_template='/schemas/{model}.json#/')`, here `{model}`
+will be replaced with the model naming using `str.format()`.
+
 ## Field customisation
 
 Optionally, the `Field` function can be used to provide extra information about the field and validations.

pydantic/main.py

@@ -31,7 +31,7 @@
 from .fields import SHAPE_MAPPING, ModelField, Undefined
 from .json import custom_pydantic_encoder, pydantic_encoder
 from .parse import Protocol, load_file, load_str_bytes
-from .schema import model_schema
+from .schema import default_ref_template, model_schema
 from .types import PyObject, StrBytes
 from .typing import AnyCallable, ForwardRef, get_origin, is_classvar, resolve_annotations, update_field_forward_refs
 from .utils import (
@@ -562,19 +562,23 @@ def copy(
         return m
 
     @classmethod
-    def schema(cls, by_alias: bool = True) -> 'DictStrAny':
-        cached = cls.__schema_cache__.get(by_alias)
+    def schema(cls, by_alias: bool = True, ref_template: str = default_ref_template) -> 'DictStrAny':
+        cached = cls.__schema_cache__.get((by_alias, ref_template))
         if cached is not None:
             return cached
-        s = model_schema(cls, by_alias=by_alias)
-        cls.__schema_cache__[by_alias] = s
+        s = model_schema(cls, by_alias=by_alias, ref_template=ref_template)
+        cls.__schema_cache__[(by_alias, ref_template)] = s
         return s
 
     @classmethod
-    def schema_json(cls, *, by_alias: bool = True, **dumps_kwargs: Any) -> str:
+    def schema_json(
+        cls, *, by_alias: bool = True, ref_template: str = default_ref_template, **dumps_kwargs: Any
+    ) -> str:
         from .json import pydantic_encoder
 
-        return cls.__config__.json_dumps(cls.schema(by_alias=by_alias), default=pydantic_encoder, **dumps_kwargs)
+        return cls.__config__.json_dumps(
+            cls.schema(by_alias=by_alias, ref_template=ref_template), default=pydantic_encoder, **dumps_kwargs
+        )
 
     @classmethod
     def __get_validators__(cls) -> 'CallableGenerator':

pydantic/schema.py

@@ -64,7 +64,7 @@
     from .main import BaseModel  # noqa: F401
 
 default_prefix = '#/definitions/'
-
+default_ref_template = '#/definitions/{model}'
 
 TypeModelOrEnum = Union[Type['BaseModel'], Type[Enum]]
 TypeModelSet = Set[TypeModelOrEnum]
@@ -77,6 +77,7 @@ def schema(
     title: Optional[str] = None,
     description: Optional[str] = None,
     ref_prefix: Optional[str] = None,
+    ref_template: str = default_ref_template,
 ) -> Dict[str, Any]:
     """
     Process a list of models and generate a single JSON Schema with all of them defined in the ``definitions``
@@ -91,11 +92,13 @@ def schema(
       else, e.g. for OpenAPI use ``#/components/schemas/``. The resulting generated schemas will still be at the
       top-level key ``definitions``, so you can extract them from there. But all the references will have the set
       prefix.
+    :param ref_template: Use a ``string.format()`` template for ``$ref`` instead of a prefix. This can be useful
+      for references that cannot be represented by ``ref_prefix`` such as a definition stored in another file. For
+      a sibling json file in a ``/schemas`` directory use ``"/schemas/${model}.json#"``.
     :return: dict with the JSON Schema with a ``definitions`` top-level key including the schema definitions for
       the models and sub-models passed in ``models``.
     """
     clean_models = [get_model(model) for model in models]
-    ref_prefix = ref_prefix or default_prefix
     flat_models = get_flat_models_from_models(clean_models)
     model_name_map = get_model_name_map(flat_models)
     definitions = {}
@@ -106,7 +109,11 @@ def schema(
         output_schema['description'] = description
     for model in clean_models:
         m_schema, m_definitions, m_nested_models = model_process_schema(
-            model, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix
+            model,
+            by_alias=by_alias,
+            model_name_map=model_name_map,
+            ref_prefix=ref_prefix,
+            ref_template=ref_template,
         )
         definitions.update(m_definitions)
         model_name = model_name_map[model]
@@ -117,7 +124,10 @@ def schema(
 
 
 def model_schema(
-    model: Union[Type['BaseModel'], Type['DataclassType']], by_alias: bool = True, ref_prefix: Optional[str] = None
+    model: Union[Type['BaseModel'], Type['DataclassType']],
+    by_alias: bool = True,
+    ref_prefix: Optional[str] = None,
+    ref_template: str = default_ref_template,
 ) -> Dict[str, Any]:
     """
     Generate a JSON Schema for one model. With all the sub-models defined in the ``definitions`` top-level
@@ -130,20 +140,22 @@ def model_schema(
       else, e.g. for OpenAPI use ``#/components/schemas/``. The resulting generated schemas will still be at the
       top-level key ``definitions``, so you can extract them from there. But all the references will have the set
       prefix.
+    :param ref_template: Use a ``string.format()`` template for ``$ref`` instead of a prefix. This can be useful for
+      references that cannot be represented by ``ref_prefix`` such as a definition stored in another file. For a
+      sibling json file in a ``/schemas`` directory use ``"/schemas/${model}.json#"``.
     :return: dict with the JSON Schema for the passed ``model``
     """
     model = get_model(model)
-    ref_prefix = ref_prefix or default_prefix
     flat_models = get_flat_models_from_model(model)
     model_name_map = get_model_name_map(flat_models)
     model_name = model_name_map[model]
     m_schema, m_definitions, nested_models = model_process_schema(
-        model, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix
+        model, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix, ref_template=ref_template
     )
     if model_name in nested_models:
         # model_name is in Nested models, it has circular references
         m_definitions[model_name] = m_schema
-        m_schema = {'$ref': ref_prefix + model_name}
+        m_schema = get_schema_ref(model_name, ref_prefix, ref_template, False)
     if m_definitions:
         m_schema.update({'definitions': m_definitions})
     return m_schema
@@ -179,6 +191,7 @@ def field_schema(
     by_alias: bool = True,
     model_name_map: Dict[TypeModelOrEnum, str],
     ref_prefix: Optional[str] = None,
+    ref_template: str = default_ref_template,
     known_models: TypeModelSet = None,
 ) -> Tuple[Dict[str, Any], Dict[str, Any], Set[str]]:
     """
@@ -192,10 +205,12 @@ def field_schema(
     :param model_name_map: used to generate the JSON Schema references to other models included in the definitions
     :param ref_prefix: the JSON Pointer prefix to use for references to other schemas, if None, the default of
       #/definitions/ will be used
+    :param ref_template: Use a ``string.format()`` template for ``$ref`` instead of a prefix. This can be useful for
+      references that cannot be represented by ``ref_prefix`` such as a definition stored in another file. For a
+      sibling json file in a ``/schemas`` directory use ``"/schemas/${model}.json#"``.
     :param known_models: used to solve circular references
     :return: tuple of the schema for this field and additional definitions
     """
-    ref_prefix = ref_prefix or default_prefix
     s, schema_overrides = get_field_info_schema(field)
 
     validation_schema = get_field_schema_validations(field)
@@ -209,6 +224,7 @@ def field_schema(
         model_name_map=model_name_map,
         schema_overrides=schema_overrides,
         ref_prefix=ref_prefix,
+        ref_template=ref_template,
         known_models=known_models or set(),
     )
     # $ref will only be returned when there are no schema_overrides
@@ -380,6 +396,7 @@ def field_type_schema(
     *,
     by_alias: bool,
     model_name_map: Dict[TypeModelOrEnum, str],
+    ref_template: str,
     schema_overrides: bool = False,
     ref_prefix: Optional[str] = None,
     known_models: TypeModelSet,
@@ -393,10 +410,14 @@ def field_type_schema(
     definitions = {}
     nested_models: Set[str] = set()
     f_schema: Dict[str, Any]
-    ref_prefix = ref_prefix or default_prefix
     if field.shape in {SHAPE_LIST, SHAPE_TUPLE_ELLIPSIS, SHAPE_SEQUENCE, SHAPE_SET, SHAPE_FROZENSET, SHAPE_ITERABLE}:
         items_schema, f_definitions, f_nested_models = field_singleton_schema(
-            field, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix, known_models=known_models
+            field,
+            by_alias=by_alias,
+            model_name_map=model_name_map,
+            ref_prefix=ref_prefix,
+            ref_template=ref_template,
+            known_models=known_models,
         )
         definitions.update(f_definitions)
         nested_models.update(f_nested_models)
@@ -409,7 +430,12 @@ def field_type_schema(
         key_field = cast(ModelField, field.key_field)
         regex = getattr(key_field.type_, 'regex', None)
         items_schema, f_definitions, f_nested_models = field_singleton_schema(
-            field, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix, known_models=known_models
+            field,
+            by_alias=by_alias,
+            model_name_map=model_name_map,
+            ref_prefix=ref_prefix,
+            ref_template=ref_template,
+            known_models=known_models,
         )
         definitions.update(f_definitions)
         nested_models.update(f_nested_models)
@@ -425,7 +451,12 @@ def field_type_schema(
         sub_fields = cast(List[ModelField], field.sub_fields)
         for sf in sub_fields:
             sf_schema, sf_definitions, sf_nested_models = field_type_schema(
-                sf, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix, known_models=known_models
+                sf,
+                by_alias=by_alias,
+                model_name_map=model_name_map,
+                ref_prefix=ref_prefix,
+                ref_template=ref_template,
+                known_models=known_models,
             )
             definitions.update(sf_definitions)
             nested_models.update(sf_nested_models)
@@ -441,6 +472,7 @@ def field_type_schema(
             model_name_map=model_name_map,
             schema_overrides=schema_overrides,
             ref_prefix=ref_prefix,
+            ref_template=ref_template,
             known_models=known_models,
         )
         definitions.update(f_definitions)
@@ -460,6 +492,7 @@ def model_process_schema(
     by_alias: bool = True,
     model_name_map: Dict[TypeModelOrEnum, str],
     ref_prefix: Optional[str] = None,
+    ref_template: str = default_ref_template,
     known_models: TypeModelSet = None,
 ) -> Tuple[Dict[str, Any], Dict[str, Any], Set[str]]:
     """
@@ -471,7 +504,6 @@ def model_process_schema(
     """
     from inspect import getdoc, signature
 
-    ref_prefix = ref_prefix or default_prefix
     known_models = known_models or set()
     if lenient_issubclass(model, Enum):
         model = cast(Type[Enum], model)
@@ -484,7 +516,12 @@ def model_process_schema(
         s['description'] = doc
     known_models.add(model)
     m_schema, m_definitions, nested_models = model_type_schema(
-        model, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix, known_models=known_models
+        model,
+        by_alias=by_alias,
+        model_name_map=model_name_map,
+        ref_prefix=ref_prefix,
+        ref_template=ref_template,
+        known_models=known_models,
     )
     s.update(m_schema)
     schema_extra = model.__config__.schema_extra
@@ -503,6 +540,7 @@ def model_type_schema(
     *,
     by_alias: bool,
     model_name_map: Dict[TypeModelOrEnum, str],
+    ref_template: str,
     ref_prefix: Optional[str] = None,
     known_models: TypeModelSet,
 ) -> Tuple[Dict[str, Any], Dict[str, Any], Set[str]]:
@@ -512,15 +550,19 @@ def model_type_schema(
     Take a single ``model`` and generate the schema for its type only, not including additional
     information as title, etc. Also return additional schema definitions, from sub-models.
     """
-    ref_prefix = ref_prefix or default_prefix
     properties = {}
     required = []
     definitions: Dict[str, Any] = {}
     nested_models: Set[str] = set()
     for k, f in model.__fields__.items():
         try:
             f_schema, f_definitions, f_nested_models = field_schema(
-                f, by_alias=by_alias, model_name_map=model_name_map, ref_prefix=ref_prefix, known_models=known_models
+                f,
+                by_alias=by_alias,
+                model_name_map=model_name_map,
+                ref_prefix=ref_prefix,
+                ref_template=ref_template,
+                known_models=known_models,
             )
         except SkipField as skip:
             warnings.warn(skip.message, UserWarning)
@@ -578,6 +620,7 @@ def field_singleton_sub_fields_schema(
     *,
     by_alias: bool,
     model_name_map: Dict[TypeModelOrEnum, str],
+    ref_template: str,
     schema_overrides: bool = False,
     ref_prefix: Optional[str] = None,
     known_models: TypeModelSet,
@@ -588,7 +631,6 @@ def field_singleton_sub_fields_schema(
     Take a list of Pydantic ``ModelField`` from the declaration of a type with parameters, and generate their
     schema. I.e., fields used as "type parameters", like ``str`` and ``int`` in ``Tuple[str, int]``.
     """
-    ref_prefix = ref_prefix or default_prefix
     definitions = {}
     nested_models: Set[str] = set()
     sub_fields = [sf for sf in sub_fields if sf.include_in_schema()]
@@ -599,6 +641,7 @@ def field_singleton_sub_fields_schema(
             model_name_map=model_name_map,
             schema_overrides=schema_overrides,
             ref_prefix=ref_prefix,
+            ref_template=ref_template,
             known_models=known_models,
         )
     else:
@@ -610,6 +653,7 @@ def field_singleton_sub_fields_schema(
                 model_name_map=model_name_map,
                 schema_overrides=schema_overrides,
                 ref_prefix=ref_prefix,
+                ref_template=ref_template,
                 known_models=known_models,
             )
             definitions.update(sub_definitions)
@@ -669,8 +713,11 @@ def add_field_type_to_schema(field_type: Any, schema: Dict[str, Any]) -> None:
             break
 
 
-def get_schema_ref(ref_name: str, schema_overrides: bool) -> Dict[str, Any]:
-    schema_ref = {'$ref': ref_name}
+def get_schema_ref(name: str, ref_prefix: Optional[str], ref_template: str, schema_overrides: bool) -> Dict[str, Any]:
+    if ref_prefix:
+        schema_ref = {'$ref': ref_prefix + name}
+    else:
+        schema_ref = {'$ref': ref_template.format(model=name)}
     return {'allOf': [schema_ref]} if schema_overrides else schema_ref
 
 
@@ -679,6 +726,7 @@ def field_singleton_schema(  # noqa: C901 (ignore complexity)
     *,
     by_alias: bool,
     model_name_map: Dict[TypeModelOrEnum, str],
+    ref_template: str,
     schema_overrides: bool = False,
     ref_prefix: Optional[str] = None,
     known_models: TypeModelSet,
@@ -690,7 +738,6 @@ def field_singleton_schema(  # noqa: C901 (ignore complexity)
     """
     from .main import BaseModel  # noqa: F811
 
-    ref_prefix = ref_prefix or default_prefix
     definitions: Dict[str, Any] = {}
     nested_models: Set[str] = set()
     if field.sub_fields:
@@ -700,6 +747,7 @@ def field_singleton_schema(  # noqa: C901 (ignore complexity)
             model_name_map=model_name_map,
             schema_overrides=schema_overrides,
             ref_prefix=ref_prefix,
+            ref_template=ref_template,
             known_models=known_models,
         )
     if field.type_ is Any or field.type_.__class__ == TypeVar:
@@ -718,6 +766,7 @@ def field_singleton_schema(  # noqa: C901 (ignore complexity)
                 by_alias=by_alias,
                 model_name_map=model_name_map,
                 ref_prefix=ref_prefix,
+                ref_template=ref_template,
                 known_models=known_models,
             )
         literal_value = values[0]
@@ -727,7 +776,7 @@ def field_singleton_schema(  # noqa: C901 (ignore complexity)
     if lenient_issubclass(field_type, Enum):
         enum_name = normalize_name(field_type.__name__)
         f_schema, schema_overrides = get_field_info_schema(field)
-        f_schema.update(get_schema_ref(ref_prefix + enum_name, schema_overrides))
+        f_schema.update(get_schema_ref(enum_name, ref_prefix, ref_template, schema_overrides))
         definitions[enum_name] = enum_process_schema(field_type)
     else:
         add_field_type_to_schema(field_type, f_schema)
@@ -751,14 +800,15 @@ def field_singleton_schema(  # noqa: C901 (ignore complexity)
                 by_alias=by_alias,
                 model_name_map=model_name_map,
                 ref_prefix=ref_prefix,
+                ref_template=ref_template,
                 known_models=known_models,
             )
             definitions.update(sub_definitions)
             definitions[model_name] = sub_schema
             nested_models.update(sub_nested_models)
         else:
             nested_models.add(model_name)
-        schema_ref = get_schema_ref(ref_prefix + model_name, schema_overrides)
+        schema_ref = get_schema_ref(model_name, ref_prefix, ref_template, schema_overrides)
         return schema_ref, definitions, nested_models
 
     raise ValueError(f'Value not declarable with JSON Schema, field: {field}')