refactor(client/raw_queries): new internal structure

m3at · Aug 4, 2024 · b45349e · b45349e
1 parent 2d3e065
commit b45349e
Show file tree

Hide file tree

Showing 2 changed files with 102 additions and 60 deletions.
diff --git a/mypy.ini b/mypy.ini
@@ -3,7  3,7 @@ pretty = True
 python_version = 3.8
 show_error_codes = True
 plugins = pydantic.mypy
-exclude = tests/integrations|prisma/_vendor
 exclude = tests/integrations|prisma/_vendor|prisma/_raw_query
 
 strict_equality = True
 implicit_reexport = True

diff --git a/src/prisma/_raw_query.py b/src/prisma/_raw_query.py
@@ -1,64  1,97 @@
 from __future__ import annotations
 
 import json
-from typing import (
-    Any,
-    Callable,
-    overload,
-)
 from typing import Any, Callable, overload
 from typing_extensions import Literal
 
 from ._types import BaseModelT
 from ._compat import model_parse
 
-# From: https://github.com/prisma/prisma/blob/main/packages/client/src/runtime/utils/deserializeRawResults.ts
-# Last checked: 2022-12-04
-"""
-type PrismaType =
-  | 'int'
-  | 'bigint'
-  | 'float'
-  | 'double'
-  | 'string'
-  | 'enum'
-  | 'bytes'
-  | 'bool'
-  | 'char'
-  | 'decimal'
-  | 'json'
-  | 'xml'
-  | 'uuid'
-  | 'datetime'
-  | 'date'
-  | 'time'
-  | 'array'
-  | 'null'
-"""
 # from https://github.com/prisma/prisma/blob/7da6f030350931eff8574e805acb9c0de9087e8e/packages/client/src/runtime/utils/deserializeRawResults.ts
 PrismaType = Literal[
     'int',
     'bigint',
     'float',
     'double',
     'string',
     'enum',
     'bytes',
     'bool',
     'char',
     'decimal',
     'json',
     'xml',
     'uuid',
     'datetime',
     'date',
     'time',
     'int-array',
     'bigint-array',
     'float-array',
     'double-array',
     'string-array',
     'enum-array',
     'bytes-array',
     'bool-array',
     'char-array',
     'decimal-array',
     'json-array',
     'xml-array',
     'uuid-array',
     'datetime-array',
     'date-array',
     'time-array',
     'unknown-array',
     'unknown',
 ]
 
 
 class RawQueryResult:
     columns: list[str]
     types: list[PrismaType]
     rows: list[list[object]]
 
     def __init__(
         self,
         *,
         columns: list[str],
         types: list[PrismaType],
         rows: list[list[object]],
     ) -> None:
         self.columns = columns
         self.types = types
         self.rows = rows
 
 
 @overload
-def deserialize_raw_results(raw_list: list[dict[str, Any]]) -> list[dict[str, Any]]: ...
 def deserialize_raw_results(raw_result: dict[str, Any]) -> list[dict[str, Any]]: ...
 
 
 @overload
 def deserialize_raw_results(
-    raw_list: list[dict[str, object]],
     raw_result: dict[str, Any],
     model: type[BaseModelT],
 ) -> list[BaseModelT]: ...
 
 
 def deserialize_raw_results(
-    raw_list: list[dict[str, Any]],
     raw_result: dict[str, Any],
     model: type[BaseModelT] | None = None,
 ) -> list[BaseModelT] | list[dict[str, Any]]:
     """Deserialize a list of raw query results into their rich Python types.
 
     If `model` is given, convert each result into the corresponding model.
     Otherwise results are returned as a dictionary
     """
     result = RawQueryResult(
         columns=raw_result['columns'],
         types=raw_result['types'],
         rows=raw_result['rows'],
     )
     if model is not None:
-        return [_deserialize_prisma_object(obj, model=model, for_model=True) for obj in raw_list]
         return [_deserialize_prisma_object(obj, result=result, model=model, for_model=True) for obj in result.rows]
 
-    return [_deserialize_prisma_object(obj, for_model=False) for obj in raw_list]
     return [_deserialize_prisma_object(obj, result=result, for_model=False) for obj in result.rows]
 
 
 # NOTE: this very weird `for_model` API is simply here as a workaround for
@@ -69,37  102,63 @@ def deserialize_raw_results(
 
 @overload
 def _deserialize_prisma_object(
-    raw_obj: dict[str, Any],
     fields: list[object],
     *,
     result: RawQueryResult,
     for_model: bool,
 ) -> dict[str, Any]: ...
 
 
 @overload
 def _deserialize_prisma_object(
-    raw_obj: dict[str, object],
     fields: list[object],
     *,
     result: RawQueryResult,
     for_model: bool,
     model: type[BaseModelT],
 ) -> BaseModelT: ...
 
 
 def _deserialize_prisma_object(
-    raw_obj: dict[Any, Any],
     fields: list[object],
     *,
     result: RawQueryResult,
     for_model: bool,
     model: type[BaseModelT] | None = None,
 ) -> BaseModelT | dict[str, Any]:
     # create a local reference to avoid performance penalty of global
     # lookups on some python versions
     _deserializers = DESERIALIZERS
 
-    new_obj = {}
-    for key, raw_value in raw_obj.items():
-        value = raw_value['prisma__value']
-        prisma_type = raw_value['prisma__type']
     new_obj: dict[str, Any] = {}
     for i, field in enumerate(fields):
         key = result.columns[i]
         prisma_type = result.types[i]
 
-        new_obj[key] = _deserializers[prisma_type](value, for_model) if prisma_type in _deserializers else value
         if prisma_type.endswith('-array'):
             if field is None:
                 # array fields can stil be `None`
                 new_obj[key] = None
                 continue
 
             if not isinstance(field, list):
                 raise TypeError(
                     f'Expected array data for {key} column with internal type {prisma_type}',
                 )
 
             item_type, _ = prisma_type.split('-')
 
             new_obj[key] = [
                 _deserializers[item_type](value, for_model)
                 #
                 if item_type in _deserializers
                 else value
                 for value in field
             ]
         else:
             value = field
 
             new_obj[key] = _deserializers[prisma_type](value, for_model) if prisma_type in _deserializers else value
 
     if model is not None:
         return model_parse(model, new_obj)
@@ -115,22  174,6 @@ def _deserialize_decimal(value: str, _for_model: bool) -> float:
     return float(value)
 
 
-def _deserialize_array(value: list[Any], for_model: bool) -> list[Any]:
-    # create a local reference to avoid performance penalty of global
-    # lookups on some python versions
-    _deserializers = DESERIALIZERS
-
-    arr = []
-    for entry in value:
-        prisma_type = entry['prisma__type']
-        prisma_value = entry['prisma__value']
-        arr.append(
-            (_deserializers[prisma_type](prisma_value, for_model) if prisma_type in _deserializers else prisma_value)
-        )
-
-    return arr
-
-
 def _deserialize_json(value: object, for_model: bool) -> object:
     # TODO: this may break if someone inserts just a string into the database
     if not isinstance(value, str) and for_model:
@@ -145,9  188,8 @@ def _deserialize_json(value: object, for_model: bool) -> object:
     return value
 
 
-DESERIALIZERS: dict[str, Callable[[Any, bool], object]] = {
 DESERIALIZERS: dict[PrismaType, Callable[[Any, bool], object]] = {
     'bigint': _deserialize_bigint,
     'decimal': _deserialize_decimal,
-    'array': _deserialize_array,
     'json': _deserialize_json,
 }