Update to supporting Pydantic V2

dandischema/consts.py

@@ -1,5 +1,14 @@
-DANDI_SCHEMA_VERSION = "0.6.4"
-ALLOWED_INPUT_SCHEMAS = ["0.4.4", "0.5.1", "0.5.2", "0.6.0", "0.6.1", "0.6.2", "0.6.3"]
+DANDI_SCHEMA_VERSION = "0.7.0"
+ALLOWED_INPUT_SCHEMAS = [
+ "0.4.4",
+ "0.5.1",
+ "0.5.2",
+ "0.6.0",
+ "0.6.1",
+ "0.6.2",
+ "0.6.3",
+ "0.6.4",
+]
 
 # ATM we allow only for a single target version which is current
 # migrate has a guard now for this since it cannot migrate to anything but current

dandischema/datacite.py

@@ -219,7 +219,7 @@
  else:
  continue
  elif rel_el.url is not None:
- ident_id = rel_el.url
+ ident_id = str(rel_el.url)
  ident_tp = "URL"
  rel_dict = {
  "relatedIdentifier": ident_id.lower(),

dandischema/digests/zarr.py

@@ -2,6 +2,7 @@
 
 from functools import total_ordering
 import hashlib
+import json
 import re
 from typing import Dict, List, Optional, Tuple
 
@@ -109,7 +110,10 @@ def aggregate_digest(self, checksums: ZarrChecksums) -> str:
  """Generate an aggregated digest for a list of ZarrChecksums."""
  # Use the most compact separators possible
  # content = json.dumps([asdict(zarr_md5) for zarr_md5 in checksums], separators=(',', ':'))0
- content = checksums.json(**ENCODING_KWARGS) # type: ignore[arg-type]
+ content = json.dumps(
+ checksums.model_dump(mode="json"),
+ **ENCODING_KWARGS, # type: ignore[arg-type]
+ )
  h = hashlib.md5()
  h.update(content.encode("utf-8"))
  md5 = h.hexdigest()
@@ -125,14 +129,17 @@ def aggregate_digest(self, checksums: ZarrChecksums) -> str:
  def serialize(self, zarr_checksum_listing: ZarrChecksumListing) -> str:
  """Serialize a ZarrChecksumListing into a string."""
  # return json.dumps(asdict(zarr_checksum_listing))
- return zarr_checksum_listing.json(**ENCODING_KWARGS) # type: ignore[arg-type]
+ return json.dumps(
+ zarr_checksum_listing.model_dump(mode="json"),
+ **ENCODING_KWARGS, # type: ignore[arg-type]
+ )
 
  def deserialize(self, json_str: str) -> ZarrChecksumListing:
  """Deserialize a string into a ZarrChecksumListing."""
  # listing = ZarrChecksumListing(**json.loads(json_str))
  # listing.checksums = [ZarrChecksum(**checksum) for checksum in listing.checksums]
  # return listing
- return ZarrChecksumListing.parse_raw(json_str)
+ return ZarrChecksumListing.model_validate_json(json_str)
 
  def generate_listing(
  self,

dandischema/metadata.py

@@ -1,7 +1,9 @@
 from copy import deepcopy
+from enum import Enum
+from inspect import isclass
 import json
 from pathlib import Path
-from typing import Any, Dict, Iterable, Optional, TypeVar, Union, cast
+from typing import Any, Dict, Iterable, Optional, TypeVar, Union, cast, get_args
 
 import jsonschema
 import pydantic
@@ -15,7 +17,12 @@
 )
 from .exceptions import JsonschemaValidationError, PydanticValidationError
 from . import models
-from .utils import _ensure_newline, version2tuple
+from .utils import (
+ TransitionalGenerateJsonSchema,
+ _ensure_newline,
+ strip_top_level_optional,
+ version2tuple,
+)
 
 schema_map = {
  "Dandiset": "dandiset.json",
@@ -56,29 +63,56 @@ def generate_context() -> dict:
  fields: Dict[str, Any] = {}
  for val in dir(models):
  klass = getattr(models, val)
- if not isinstance(klass, pydantic.main.ModelMetaclass):
+ if not isclass(klass) or not issubclass(klass, pydantic.BaseModel):
  continue
  if hasattr(klass, "_ldmeta"):
- if "nskey" in klass._ldmeta:
+ if "nskey" in klass._ldmeta.default:
  name = klass.__name__
- fields[name] = f'{klass._ldmeta["nskey"]}:{name}'
- for name, field in klass.__fields__.items():
+ fields[name] = f'{klass._ldmeta.default["nskey"]}:{name}'
+ for name, field in klass.model_fields.items():
  if name == "id":
  fields[name] = "@id"
  elif name == "schemaKey":
  fields[name] = "@type"
  elif name == "digest":
  fields[name] = "@nest"
  elif name not in fields:
- if "nskey" in field.field_info.extra:
- fields[name] = {"@id": field.field_info.extra["nskey"] + ":" + name}
+ if (
+ isinstance(field.json_schema_extra, dict)
+ and "nskey" in field.json_schema_extra
+ ):
+ fields[name] = {
+ "@id": cast(str, field.json_schema_extra["nskey"]) + ":" + name
+ }
  else:
  fields[name] = {"@id": "dandi:" + name}
- if "List" in str(field.outer_type_):
+
+ # The annotation without the top-level optional
+ stripped_annotation = strip_top_level_optional(field.annotation)
+
+ # Using stringification to detect present of list in annotation is not
+ # ideal, but it works for now. A better solution should be used in the
+ # future.
+ if "list" in str(stripped_annotation).lower():
  fields[name]["@container"] = "@set"
+
+ # Handle the case where the type of the element of a list is
+ # an Enum type
+ type_args = get_args(stripped_annotation)
+ if (
+ len(type_args) == 1
+ and isclass(type_args[0])
+ and issubclass(type_args[0], Enum)
+ ):
+ fields[name]["@type"] = "@id"
+
  if name == "contributor":
  fields[name]["@container"] = "@list"
- if "enum" in str(field.type_) or name in ["url", "hasMember"]:
+ if (
+ isclass(stripped_annotation)
+ and issubclass(stripped_annotation, Enum)
+ or name in ["url", "hasMember"]
+ ):
  fields[name]["@type"] = "@id"
 
  for item in models.DigestType:
@@ -96,7 +130,14 @@ def publish_model_schemata(releasedir: Union[str, Path]) -> Path:
  vdir.mkdir(exist_ok=True, parents=True)
  for class_, filename in schema_map.items():
  (vdir / filename).write_text(
- _ensure_newline(getattr(models, class_).schema_json(indent=2))
+ _ensure_newline(
+ json.dumps(
+ getattr(models, class_).model_json_schema(
+ schema_generator=TransitionalGenerateJsonSchema
+ ),
+ indent=2,
+ )
+ )
  )
  (vdir / "context.json").write_text(
  _ensure_newline(json.dumps(generate_context(), indent=2))
@@ -177,7 +218,9 @@ def validate(
  if json_validation:
  if schema_version == DANDI_SCHEMA_VERSION:
  klass = getattr(models, schema_key)
- schema = klass.schema()
+ schema = klass.model_json_schema(
+ schema_generator=TransitionalGenerateJsonSchema
+ )
  else:
  if schema_key not in schema_map:
  raise ValueError(
@@ -196,7 +239,7 @@ def validate(
  except pydantic.ValidationError as exc:
  messages = []
  for el in exc.errors():
- if not missing_ok or el["msg"] != "field required":
+ if not missing_ok or el["type"] != "missing":
  messages.append(el)
  if messages:
  raise PydanticValidationError(messages) # type: ignore[arg-type]
@@ -245,6 +288,8 @@ def migrate(
  if "schemaKey" not in obj:
  obj["schemaKey"] = "Dandiset"
  obj["schemaVersion"] = to_version
+ if version2tuple(schema_version) < version2tuple("0.7.0"):
+ pass # TODO: do necessary changes so that old metadata (before 0.7.0) looks like new (0.7.0)
  return obj
 
 
@@ -346,4 +391,4 @@ def aggregate_assets_summary(metadata: Iterable[Dict[str, Any]]) -> dict:
  len(stats.pop("tissuesample", [])) + len(stats.pop("slice", []))
  ) or None
  stats["numberOfCells"] = len(stats.pop("cell", [])) or None
- return models.AssetsSummary(**stats).json_dict()
+ return models.AssetsSummary(**stats).model_dump(mode="json", exclude_none=True)