Repository: yukinarit/pyserde
Pull request #339 (2023-04-21 14:04:39)

Union directly
This example works correctly now

```python
@serde
@dataclass
class Foo:
    a: int

@serde
@dataclass
class Bar:
    a: int

bar = Bar(10)
s = to_json(bar)
print(s)
print(from_json(Union[Foo, Bar], s))
```

However, This will introduce a breaking change! The default
behaviour when you pass Union directly was "Untagged", but since
v0.12.0 it is "ExternalTagging".

The following code prints `{"a": 10}` until v0.11.1, but prints `{"Bar": {"a": 10}}` since v0.12.0
```python
print(to_json(bar))
```


codecov[bot]: ## [Codecov](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit) Report
Patch coverage: **`79.62`**% and project coverage change: **`-0.38`** :warning:
> Comparison is base [(`35f3f93`)](https://codecov.io/gh/yukinarit/pyserde/commit/35f3f93adf32b5e64fd43ab5f0806ecbe1239543?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit) 91.09% compared to head [(`a7e8ed5`)](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit) 90.72%.

> :exclamation: Current head a7e8ed5 differs from pull request most recent head f431a00. Consider uploading reports for the commit f431a00 to get more accurate results

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #339      +/-   ##
==========================================
- Coverage   91.09%   90.72%   -0.38%     
==========================================
  Files          12       12              
  Lines        1775     1800      +25     
  Branches      376      379       +3     
==========================================
+ Hits         1617     1633      +16     
- Misses        110      118       +8     
- Partials       48       49       +1     
```


| [Impacted Files](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit) | Coverage Δ | |
|---|---|---|
| [serde/inspect.py](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit#diff-c2VyZGUvaW5zcGVjdC5weQ==) | `46.66% <50.00%> (ø)` | |
| [serde/se.py](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit#diff-c2VyZGUvc2UucHk=) | `96.50% <61.53%> (-1.56%)` | :arrow_down: |
| [serde/core.py](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit#diff-c2VyZGUvY29yZS5weQ==) | `90.90% <85.71%> (-0.48%)` | :arrow_down: |
| [serde/de.py](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit#diff-c2VyZGUvZGUucHk=) | `96.44% <87.50%> (-0.04%)` | :arrow_down: |
| [serde/json.py](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit#diff-c2VyZGUvanNvbi5weQ==) | `100.00% <100.00%> (ø)` | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit)

</details>

[:umbrella: View full report in Codecov by Sentry](https://codecov.io/gh/yukinarit/pyserde/pull/339?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=yukinarit).


Patch:
diff --git a/docs/en/union.md b/docs/en/union.md
index e246835..a0c6d10 100644
--- a/docs/en/union.md
+++ b/docs/en/union.md
@@ -66,3 +66,84 @@ A class declaration with `AdjacentTagging` looks like below. If you serialize `F
 class Foo:
     a: Union[Bar, Baz]
 ```
+
+## (de)serializing Union types directly
+
+New in v0.12.0.
+
+Passing Union types directly in (de)serialize APIs (e.g. to_json, from_json) was partially supported prior to v0.12, but the union type was always treated as untagged. Users had no way to change the union tagging. The following example code wasn't able to correctly deserialize into `Bar` due to untagged.
+
+```python
+@serde
+@dataclass
+class Foo:
+    a: int
+
+@serde
+@dataclass
+class Bar:
+    a: int
+
+bar = Bar(10)
+s = to_json(bar)
+print(s)
+# prints {"a": 10}
+print(from_json(Union[Foo, Bar], s))
+# prints Foo(10)
+```
+
+Since v0.12.0, pyserde can handle union that's passed in (de)serialize APIs a bit nicely. The union type is treated as externally tagged as that is the default tagging in pyserde. So the above example can correctly (de)serialize as `Bar`.
+
+```python
+@serde
+@dataclass
+class Foo:
+    a: int
+
+@serde
+@dataclass
+class Bar:
+    a: int
+
+bar = Bar(10)
+s = to_json(bar)
+print(s)
+# prints {"Bar" {"a": 10}}
+print(from_json(Union[Foo, Bar], s))
+# prints Bar(10)
+```
+
+Also you can change the tagging using `serde.InternalTagging`, `serde.AdjacentTagging` and `serde.Untagged`.
+
+Now try to change the tagging for the above example. You need to pass a new argument `cls` in `to_json`. Also union class must be wrapped in either `InternalTagging`, `AdjacentTaging` or `Untagged` with required parameters.
+
+* InternalTagging
+    ```python
+    from serde import InternalTagging
+
+    s = to_json(bar, cls=InternalTagging("type", Union[Foo, Bar]))
+    print(s)
+    # prints {"type": "Bar", "a": 10}
+    print(from_json(InternalTagging("type", Union[Foo, Bar]), s))
+    # prints Bar(10)
+    ```
+* AdjacentTagging
+    ```python
+    from serde import AdjacentTagging
+
+    s = to_json(bar, cls=AdjacentTagging("type", "content", Union[Foo, Bar]))
+    print(s)
+    # prints {"type": "Bar", "content": {"a": 10}}
+    print(from_json(AdjacentTagging("type", "content", Union[Foo, Bar]), s))
+    # prints Bar(10)
+    ```
+* Untagged
+    ```python
+    from serde import Untagged
+
+    s = to_json(bar, cls=Untagged(Union[Foo, Bar]))
+    print(s)
+    # prints {"a": 10}
+    print(from_json(Untagged(Union[Foo, Bar]), s))
+    # prints Foo(10)
+    ```
diff --git a/examples/runner.py b/examples/runner.py
index 781e534..9c9c717 100644
--- a/examples/runner.py
+++ b/examples/runner.py
@@ -40,6 +40,7 @@ import type_datetime
 import type_decimal
 import union
 import union_tagging
+import union_directly
 import user_exception
 import variable_length_tuple
 import yamlfile
@@ -73,6 +74,7 @@ def run_all() -> None:
     run(type_decimal)
     run(type_datetime)
     run(union_tagging)
+    run(union_directly)
     run(generics)
     run(generics_nested)
     run(nested)
diff --git a/examples/union_directly.py b/examples/union_directly.py
new file mode 100644
index 0000000..2cca285
--- /dev/null
+++ b/examples/union_directly.py
@@ -0,0 +1,45 @@
+from dataclasses import dataclass
+from typing import Union
+
+from serde import serde, Untagged, AdjacentTagging, InternalTagging
+from serde.json import from_json, to_json
+
+
+@serde
+@dataclass
+class Bar:
+    b: int
+
+
+@serde
+@dataclass(unsafe_hash=True)
+class Baz:
+    b: int
+
+
+def main() -> None:
+    baz = Baz(10)
+
+    print("# external tagging (default)")
+    a = to_json(baz, cls=Union[Bar, Baz])
+    print("IntoJSON", a)
+    print("FromJSON", from_json(Union[Bar, Baz], a))
+
+    print("# internal tagging")
+    a = to_json(baz, cls=InternalTagging("type", Union[Bar, Baz]))
+    print("IntoJSON", a)
+    print("FromJSON", from_json(InternalTagging("type", Union[Bar, Baz]), a))
+
+    print("# adjacent tagging")
+    a = to_json(baz, cls=AdjacentTagging("type", "content", Union[Bar, Baz]))
+    print("IntoJSON", a)
+    print("FromJSON", from_json(AdjacentTagging("type", "content", Union[Bar, Baz]), a))
+
+    print("# untagged")
+    a = to_json(baz, cls=Untagged(Union[Bar, Baz]))
+    print("IntoJSON", a)
+    print("FromJSON", from_json(Untagged(Union[Bar, Baz]), a))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/serde/compat.py b/serde/compat.py
index 91ffbb8..9d5200c 100644
--- a/serde/compat.py
+++ b/serde/compat.py
@@ -114,6 +114,18 @@ DateTimeTypes = (datetime.date, datetime.time, datetime.datetime)
 """ List of datetime types """
 
 
+@dataclasses.dataclass
+class _WithTagging(Generic[T]):
+    """
+    Intermediate data structure for (de)serializaing Union without dataclass.
+    """
+
+    inner: T
+    """ Union type .e.g Union[Foo,Bar] passed in from_obj. """
+    tagging: Any
+    """ Union Tagging """
+
+
 class SerdeError(Exception):
     """
     Serde error class.
@@ -475,6 +487,13 @@ def is_union(typ: Any) -> bool:
     True
     """
 
+    try:
+        # When `_WithTagging` is received, it will check inner type.
+        if isinstance(typ, _WithTagging):
+            return is_union(typ.inner)
+    except Exception:
+        pass
+
     # Python 3.10 Union operator e.g. str | int
     if sys.version_info[:2] >= (3, 10):
         try:
diff --git a/serde/core.py b/serde/core.py
index 5ebf272..788b00a 100644
--- a/serde/core.py
+++ b/serde/core.py
@@ -8,7 +8,19 @@ import functools
 import logging
 import re
 from dataclasses import dataclass
-from typing import Any, Callable, Dict, List, Mapping, Optional, Union, Generic, TypeVar
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    List,
+    Mapping,
+    Optional,
+    Union,
+    Generic,
+    TypeVar,
+    Tuple,
+    overload,
+)
 
 import casefy
 import jinja2
@@ -36,11 +48,12 @@ from .compat import (
     is_variable_tuple,
     type_args,
     typename,
+    _WithTagging,
 )
 from .numpy import is_numpy_available, is_numpy_type
 
 __all__ = [
-    "SerdeScope",
+    "Scope",
     "gen",
     "add_func",
     "Func",
@@ -78,7 +91,148 @@ def init(debug: bool = False) -> None:
 
 
 @dataclass
-class SerdeScope:
+class Cache:
+    """
+    Cache the generated code for non-dataclass classes.
+
+    for example, a type not bound in a dataclass is passed in from_json
+
+    ```
+    from_json(Union[Foo, Bar], ...)
+    ```
+
+    It creates the following wrapper dataclass on the fly,
+
+    ```
+    @serde
+    @dataclass
+    class Union_Foo_bar:
+        v: Union[Foo, Bar]
+    ```
+
+    Then store this class in this cache. Whenever the same type is passed,
+    the class is retrieved from this cache. So the overhead of the codegen
+    should be only once.
+    """
+
+    classes: Dict[str, Type[Any]] = dataclasses.field(default_factory=dict)
+
+    def _get_class(self, cls: Type[Any]) -> Type[Any]:
+        """
+        Get a wrapper class from the the cache. If not found, it will generate
+        the class and store it in the cache.
+        """
+        class_name = f"Wrapper{typename(cls)}"
+        wrapper = self.classes.get(class_name)
+        return wrapper or self._generate_class(cls)
+
+    def _generate_class(self, cls: Type[Any]) -> Type[Any]:
+        """
+        Generate a wrapper dataclass then make the it (de)serializable using
+        @serde decorator.
+        """
+        from . import serde
+
+        class_name = f"Wrapper{typename(cls)}"
+        logger.debug(f"Generating a wrapper class code for {class_name}")
+
+        wrapper = dataclasses.make_dataclass(class_name, [("v", cls)])
+
+        serde(wrapper)
+        self.classes[class_name] = wrapper
+
+        logger.debug(f"(de)serializing code for {class_name} was generated")
+        return wrapper
+
+    def serialize(self, cls: Type[Any], obj: Any) -> Any:
+        """
+        Serialize the specified type of object into dict or tuple.
+        """
+        wrapper = self._get_class(cls)
+        scope: Scope = getattr(wrapper, SERDE_SCOPE)
+        data = scope.funcs[TO_DICT](wrapper(obj), reuse_instances=False, convert_sets=True)
+
+        logging.debug(f"Intermediate value: {data}")
+
+        return data["v"]
+
+    def deserialize(self, cls: Type[T], obj: Any) -> T:
+        """
+        Deserialize from dict or tuple into the specified type.
+        """
+        wrapper = self._get_class(cls)
+        scope: Scope = getattr(wrapper, SERDE_SCOPE)
+        return scope.funcs[FROM_DICT](data={"v": obj}).v  # type: ignore
+
+    def _get_union_class(self, cls: Type[Any]) -> Optional[Type[Any]]:
+        """
+        Get a wrapper class from the the cache. If not found, it will generate
+        the class and store it in the cache.
+        """
+        union_cls, tagging = _extract_from_with_tagging(cls)
+        class_name = union_func_name(
+            f"{tagging.produce_unique_class_name()}Union", list(type_args(union_cls))
+        )
+        wrapper = self.classes.get(class_name)
+        return wrapper or self._generate_union_class(cls)
+
+    def _generate_union_class(self, cls: Type[Any]) -> Type[Any]:
+        """
+        Generate a wrapper dataclass then make the it (de)serializable using
+        @serde decorator.
+        """
+        import serde
+
+        union_cls, tagging = _extract_from_with_tagging(cls)
+        class_name = union_func_name(
+            f"{tagging.produce_unique_class_name()}Union", list(type_args(union_cls))
+        )
+        wrapper = dataclasses.make_dataclass(class_name, [("v", union_cls)])
+        serde.serde(wrapper, tagging=tagging)
+        self.classes[class_name] = wrapper
+        return wrapper
+
+    def serialize_union(self, cls: Type[Any], obj: Any) -> Any:
+        """
+        Serialize the specified Union into dict or tuple.
+        """
+        union_cls, _ = _extract_from_with_tagging(cls)
+        wrapper = self._get_union_class(cls)
+        scope: Scope = getattr(wrapper, SERDE_SCOPE)
+        func_name = union_func_name(UNION_SE_PREFIX, list(type_args(union_cls)))
+        return scope.funcs[func_name](obj, False, False)
+
+    def deserialize_union(self, cls: Type[T], data: Any) -> T:
+        """
+        Deserialize from dict or tuple into the specified Union.
+        """
+        union_cls, _ = _extract_from_with_tagging(cls)
+        wrapper = self._get_union_class(cls)
+        scope: Scope = getattr(wrapper, SERDE_SCOPE)
+        func_name = union_func_name(UNION_DE_PREFIX, list(type_args(union_cls)))
+        return scope.funcs[func_name](cls=union_cls, data=data)  # type: ignore
+
+
+def _extract_from_with_tagging(maybe_with_tagging: Any) -> Tuple[Any, Tagging]:
+    try:
+        if isinstance(maybe_with_tagging, _WithTagging):
+            union_cls = maybe_with_tagging.inner
+            tagging = maybe_with_tagging.tagging
+        else:
+            raise Exception()
+    except Exception:
+        union_cls = maybe_with_tagging
+        tagging = ExternalTagging
+
+    return (union_cls, tagging)
+
+
+CACHE = Cache()
+""" Global cache variable for non-dataclass classes """
+
+
+@dataclass
+class Scope:
     """
     Container to store types and functions used in code generation context.
     """
@@ -173,13 +327,11 @@ def gen(
     return code
 
 
-def add_func(
-    serde_scope: SerdeScope, func_name: str, func_code: str, globals: Dict[str, Any]
-) -> None:
+def add_func(serde_scope: Scope, func_name: str, func_code: str, globals: Dict[str, Any]) -> None:
     """
-    Generate a function and add it to a SerdeScope's `funcs` dictionary.
+    Generate a function and add it to a Scope's `funcs` dictionary.
 
-    * `serde_scope`: the SerdeScope instance to modify
+    * `serde_scope`: the Scope instance to modify
     * `func_name`: the name of the function
     * `func_code`: the source code of the function
     * `globals`: global variables that should be accessible to the generated function
@@ -198,7 +350,7 @@ def is_instance(obj: Any, typ: Type[Any]) -> bool:
     Subscripted Generics e.g. `List[int]`.
     """
     if dataclasses.is_dataclass(typ):
-        serde_scope: Optional[SerdeScope] = getattr(typ, SERDE_SCOPE, None)
+        serde_scope: Optional[Scope] = getattr(typ, SERDE_SCOPE, None)
         if serde_scope:
             try:
                 serde_scope.funcs[TYPE_CHECK](obj)
@@ -623,15 +775,74 @@ class Tagging:
         if self.is_adjacent() and (self.tag is None or self.content is None):
             raise SerdeError('"tag" and "content" must be specified in AdjacentTagging')
 
+    def produce_unique_class_name(self) -> str:
+        """
+        Produce a unique class name for this tagging. The name is used for generated
+        wrapper dataclass and stored in `Cache`.
+        """
+        if self.is_internal():
+            tag = casefy.pascalcase(self.tag)
+            if not tag:
+                raise SerdeError('"tag" must be specified in InternalTagging')
+            return f"Internal{tag}"
+        elif self.is_adjacent():
+            tag = casefy.pascalcase(self.tag)
+            content = casefy.pascalcase(self.content)
+            if not tag:
+                raise SerdeError('"tag" must be specified in AdjacentTagging')
+            if not content:
+                raise SerdeError('"content" must be specified in AdjacentTagging')
+            return f"Adjacent{tag}{content}"
+        else:
+            return self.kind.name
+
+    def __call__(self, cls: T) -> _WithTagging[T]:
+        return _WithTagging(cls, self)
+
+
+@overload
+def InternalTagging(tag: str) -> Tagging:
+    ...
+
+
+@overload
+def InternalTagging(tag: str, cls: T) -> _WithTagging[T]:
+    ...
+
+
+def InternalTagging(tag: str, cls: Optional[T] = None) -> Union[Tagging, _WithTagging[T]]:
+    tagging = Tagging(tag, kind=Tagging.Kind.Internal)
+    if cls:
+        return tagging(cls)
+    else:
+        return tagging
+
 
-ExternalTagging = Tagging()
+@overload
+def AdjacentTagging(tag: str, content: str) -> Tagging:
+    ...
 
-InternalTagging = functools.partial(Tagging, kind=Tagging.Kind.Internal)
 
-AdjacentTagging = functools.partial(Tagging, kind=Tagging.Kind.Adjacent)
+@overload
+def AdjacentTagging(tag: str, content: str, cls: T) -> _WithTagging[T]:
+    ...
+
+
+def AdjacentTagging(
+    tag: str, content: str, cls: Optional[T] = None
+) -> Union[Tagging, _WithTagging[T]]:
+    tagging = Tagging(tag, content, kind=Tagging.Kind.Adjacent)
+    if cls:
+        return tagging(cls)
+    else:
+        return tagging
+
+
+ExternalTagging = Tagging()
 
 Untagged = Tagging(kind=Tagging.Kind.Untagged)
 
+
 DefaultTagging = ExternalTagging
 
 
diff --git a/serde/de.py b/serde/de.py
index 1cbe452..37ae32c 100644
--- a/serde/de.py
+++ b/serde/de.py
@@ -59,12 +59,13 @@ from .core import (
     FROM_DICT,
     FROM_ITER,
     SERDE_SCOPE,
+    CACHE,
     TYPE_CHECK,
     UNION_DE_PREFIX,
     DefaultTagging,
     Field,
     NoCheck,
-    SerdeScope,
+    Scope,
     Tagging,
     TypeCheck,
     add_func,
@@ -82,8 +83,8 @@ from .core import (
 )
 from .numpy import (
     deserialize_numpy_array,
-    deserialize_numpy_array_direct,
     deserialize_numpy_scalar,
+    deserialize_numpy_array_direct,
     is_numpy_array,
     is_numpy_scalar,
 )
@@ -214,9 +215,9 @@ def deserialize(
         # Create a scope storage used by serde.
         # Each class should get own scope. Child classes can not share scope with parent class.
         # That's why we need the "scope.cls is not cls" check.
-        scope: Optional[SerdeScope] = getattr(cls, SERDE_SCOPE, None)
+        scope: Optional[Scope] = getattr(cls, SERDE_SCOPE, None)
         if scope is None or scope.cls is not cls:
-            scope = SerdeScope(cls, reuse_instances_default=reuse_instances_default)
+            scope = Scope(cls, reuse_instances_default=reuse_instances_default)
             setattr(cls, SERDE_SCOPE, scope)
 
         # Set some globals for all generated functions
@@ -331,7 +332,7 @@ def is_dataclass_without_de(cls: Type[Any]) -> bool:
         return False
     if not hasattr(cls, SERDE_SCOPE):
         return True
-    scope: Optional[SerdeScope] = getattr(cls, SERDE_SCOPE)
+    scope: Optional[Scope] = getattr(cls, SERDE_SCOPE)
     return FROM_DICT not in scope.funcs
 
 
@@ -362,13 +363,19 @@ def from_obj(c: Type[T], o: Any, named: bool, reuse_instances: bool) -> T:
     """
 
     def deserializable_to_obj(cls):
-        serde_scope: SerdeScope = getattr(cls, SERDE_SCOPE)
+        serde_scope: Scope = getattr(cls, SERDE_SCOPE)
         func_name = FROM_DICT if named else FROM_ITER
         res = serde_scope.funcs[func_name](
             cls, maybe_generic=maybe_generic, data=o, reuse_instances=reuse_instances
         )
         return res
 
+    if is_union(c) and not is_opt(c):
+        # If a class in the argument is a non-dataclass class e.g. Union[Foo, Bar],
+        # pyserde generates a wrapper (de)serializable dataclass on the fly,
+        # and use it to deserialize into the object.
+        return CACHE.deserialize_union(c, o)
+
     if is_generic(c):
         # Store subscripted generic type such as Foo[Bar] in "maybe_generic",
         # and store origin type such as Foo in "c". Since subscripted generics
@@ -390,15 +397,6 @@ def from_obj(c: Type[T], o: Any, named: bool, reuse_instances: bool) -> T:
                 return None
             else:
                 return thisfunc(type_args(c)[0], o)
-        elif is_union(c):
-            v = None
-            for typ in type_args(c):
-                try:
-                    v = thisfunc(typ, o)
-                    break
-                except (SerdeError, ValueError):
-                    pass
-            return v
         elif is_list(c):
             if is_bare_list(c):
                 return list(o)
diff --git a/serde/json.py b/serde/json.py
index d3e5906..88eb694 100644
--- a/serde/json.py
+++ b/serde/json.py
@@ -1,7 +1,7 @@
 """
 Serialize and Deserialize in JSON format.
 """
-from typing import Any, overload
+from typing import Any, overload, Optional
 
 from typing_extensions import Type
 
@@ -51,7 +51,9 @@ class JsonDeserializer(Deserializer[str]):
         return json_loads(data, **opts)
 
 
-def to_json(obj: Any, se: Type[Serializer[str]] = JsonSerializer, **opts: Any) -> str:
+def to_json(
+    obj: Any, cls: Optional[Any] = None, se: Type[Serializer[str]] = JsonSerializer, **opts: Any
+) -> str:
     """
     Serialize the object into JSON str. [orjson](https://github.com/ijl/orjson)
     will be used if installed.
@@ -65,7 +67,7 @@ def to_json(obj: Any, se: Type[Serializer[str]] = JsonSerializer, **opts: Any) -
     If you want to use another json package, you can subclass `JsonSerializer` and implement
     your own logic.
     """
-    return se.serialize(to_dict(obj, reuse_instances=False, convert_sets=True), **opts)
+    return se.serialize(to_dict(obj, c=cls, reuse_instances=False, convert_sets=True), **opts)
 
 
 @overload
diff --git a/serde/msgpack.py b/serde/msgpack.py
index 86527f7..eadc1a0 100644
--- a/serde/msgpack.py
+++ b/serde/msgpack.py
@@ -40,6 +40,7 @@ class MsgPackDeserializer(Deserializer[bytes]):
 
 def to_msgpack(
     obj: Any,
+    cls: Optional[Any] = None,
     se: Type[Serializer[bytes]] = MsgPackSerializer,
     named: bool = True,
     ext_dict: Optional[Dict[Type[Any], int]] = None,
@@ -66,9 +67,10 @@ def to_msgpack(
         if ext_type_code is None:
             raise SerdeError(f"Could not find type code for {obj_type.__name__} in ext_dict")
 
-    to_func = to_dict if named else to_tuple
+    kwargs: Any = {"c": cls, "reuse_instances": False, "convert_sets": True}
+    dict_or_tuple = to_dict(obj, **kwargs) if named else to_tuple(obj, **kwargs)
     return se.serialize(
-        to_func(obj, reuse_instances=False, convert_sets=True),
+        dict_or_tuple,
         ext_type_code=ext_type_code,
         **opts,
     )
diff --git a/serde/pickle.py b/serde/pickle.py
index b39fec3..f36cf77 100644
--- a/serde/pickle.py
+++ b/serde/pickle.py
@@ -2,7 +2,7 @@
 Serialize and Deserialize in Pickle format.
 """
 import pickle
-from typing import Type, Any, overload
+from typing import Type, Any, overload, Optional
 
 from .compat import T
 from .de import Deserializer, from_dict
@@ -23,8 +23,10 @@ class PickleDeserializer(Deserializer[bytes]):
         return pickle.loads(data, **opts)
 
 
-def to_pickle(obj: Any, se: Type[Serializer[bytes]] = PickleSerializer, **opts: Any) -> bytes:
-    return se.serialize(to_dict(obj, reuse_instances=False), **opts)
+def to_pickle(
+    obj: Any, cls: Optional[Any] = None, se: Type[Serializer[bytes]] = PickleSerializer, **opts: Any
+) -> bytes:
+    return se.serialize(to_dict(obj, c=cls, reuse_instances=False), **opts)
 
 
 @overload
diff --git a/serde/se.py b/serde/se.py
index f8a9590..409c7ca 100644
--- a/serde/se.py
+++ b/serde/se.py
@@ -50,6 +50,7 @@ from .compat import (
     typename,
 )
 from .core import (
+    CACHE,
     SERDE_SCOPE,
     TO_DICT,
     TO_ITER,
@@ -58,7 +59,7 @@ from .core import (
     DefaultTagging,
     Field,
     NoCheck,
-    SerdeScope,
+    Scope,
     Tagging,
     TypeCheck,
     add_func,
@@ -195,9 +196,9 @@ def serialize(
         # Create a scope storage used by serde.
         # Each class should get own scope. Child classes can not share scope with parent class.
         # That's why we need the "scope.cls is not cls" check.
-        scope: Optional[SerdeScope] = getattr(cls, SERDE_SCOPE, None)
+        scope: Optional[Scope] = getattr(cls, SERDE_SCOPE, None)
         if scope is None or scope.cls is not cls:
-            scope = SerdeScope(
+            scope = Scope(
                 cls,
                 reuse_instances_default=reuse_instances_default,
                 convert_sets_default=convert_sets_default,
@@ -303,15 +304,13 @@ def is_dataclass_without_se(cls: Type[Any]) -> bool:
         return False
     if not hasattr(cls, SERDE_SCOPE):
         return True
-    scope: Optional[SerdeScope] = getattr(cls, SERDE_SCOPE)
+    scope: Optional[Scope] = getattr(cls, SERDE_SCOPE)
     return TO_DICT not in scope.funcs
 
 
-def to_obj(
-    o, named: bool, reuse_instances: bool, convert_sets: bool, c: Optional[Type[Any]] = None
-):
+def to_obj(o, named: bool, reuse_instances: bool, convert_sets: bool, c: Optional[Any] = None):
     def serializable_to_obj(object):
-        serde_scope: SerdeScope = getattr(object, SERDE_SCOPE)
+        serde_scope: Scope = getattr(object, SERDE_SCOPE)
         func_name = TO_DICT if named else TO_ITER
         return serde_scope.funcs[func_name](
             object, reuse_instances=reuse_instances, convert_sets=convert_sets
@@ -324,6 +323,13 @@ def to_obj(
             reuse_instances=reuse_instances,
             convert_sets=convert_sets,
         )
+
+        # If a class in the argument is a non-dataclass class e.g. Union[Foo, Bar],
+        # pyserde generates a wrapper (de)serializable dataclass on the fly,
+        # and use it to serialize the object.
+        if c and is_union(c) and not is_opt(c):
+            return CACHE.serialize_union(c, o)
+
         if o is None:
             return None
         if is_dataclass_without_se(o):
@@ -357,7 +363,9 @@ def astuple(v: Any) -> Tuple[Any, ...]:
     return to_tuple(v, reuse_instances=False, convert_sets=False)
 
 
-def to_tuple(o, reuse_instances: bool = ..., convert_sets: bool = ...) -> Tuple[Any, ...]:
+def to_tuple(
+    o: Any, c: Optional[Type[Any]] = None, reuse_instances: bool = ..., convert_sets: bool = ...
+) -> Tuple[Any, ...]:
     """
     Serialize object into tuple.
 
@@ -377,7 +385,7 @@ def to_tuple(o, reuse_instances: bool = ..., convert_sets: bool = ...) -> Tuple[
     >>> to_tuple(lst)
     [(10, 'foo', 100.0, True), (20, 'foo', 100.0, True)]
     """
-    return to_obj(o, named=False, reuse_instances=reuse_instances, convert_sets=convert_sets)
+    return to_obj(o, named=False, c=c, reuse_instances=reuse_instances, convert_sets=convert_sets)
 
 
 def asdict(v: Any) -> Dict[Any, Any]:
@@ -387,7 +395,9 @@ def asdict(v: Any) -> Dict[Any, Any]:
     return to_dict(v, reuse_instances=False, convert_sets=False)
 
 
-def to_dict(o, reuse_instances: bool = ..., convert_sets: bool = ...) -> Dict[Any, Any]:
+def to_dict(
+    o: Any, c: Optional[Type[Any]] = None, reuse_instances: bool = ..., convert_sets: bool = ...
+) -> Dict[Any, Any]:
     """
     Serialize object into dictionary.
 
@@ -407,7 +417,7 @@ def to_dict(o, reuse_instances: bool = ..., convert_sets: bool = ...) -> Dict[An
     >>> to_dict(lst)
     [{'i': 10, 's': 'foo', 'f': 100.0, 'b': True}, {'i': 20, 's': 'foo', 'f': 100.0, 'b': True}]
     """
-    return to_obj(o, named=True, reuse_instances=reuse_instances, convert_sets=convert_sets)
+    return to_obj(o, named=True, c=c, reuse_instances=reuse_instances, convert_sets=convert_sets)
 
 
 @dataclass
@@ -702,7 +712,7 @@ convert_sets=convert_sets), coerce(int, foo[2]),)"
         elif is_none(arg.type):
             res = "None"
         elif is_any(arg.type) or isinstance(arg.type, TypeVar):
-            res = f"to_obj({arg.varname}, True, False, False)"
+            res = f"to_obj({arg.varname}, True, False, False, c=typing.Any)"
         elif is_generic(arg.type):
             origin = get_origin(arg.type)
             assert origin
diff --git a/serde/toml.py b/serde/toml.py
index f73e922..835efe7 100644
--- a/serde/toml.py
+++ b/serde/toml.py
@@ -4,7 +4,7 @@ Serialize and Deserialize in TOML format. This module depends on
 [tomli-w](https://github.com/hukkin/tomli-w) packages.
 """
 import sys
-from typing import Type, Any, overload
+from typing import Type, Any, overload, Optional
 
 import tomli_w
 
@@ -33,7 +33,9 @@ class TomlDeserializer(Deserializer[str]):
         return tomllib.loads(data, **opts)
 
 
-def to_toml(obj: Any, se: Type[Serializer[str]] = TomlSerializer, **opts: Any) -> str:
+def to_toml(
+    obj: Any, cls: Optional[Any] = None, se: Type[Serializer[str]] = TomlSerializer, **opts: Any
+) -> str:
     """
     Serialize the object into TOML.
 
@@ -43,7 +45,7 @@ def to_toml(obj: Any, se: Type[Serializer[str]] = TomlSerializer, **opts: Any) -
     If you want to use the other toml package, you can subclass `TomlSerializer` and implement
     your own logic.
     """
-    return se.serialize(to_dict(obj, reuse_instances=False), **opts)
+    return se.serialize(to_dict(obj, c=cls, reuse_instances=False), **opts)
 
 
 @overload
diff --git a/serde/yaml.py b/serde/yaml.py
index 0977323..e672449 100644
--- a/serde/yaml.py
+++ b/serde/yaml.py
@@ -2,7 +2,7 @@
 Serialize and Deserialize in YAML format. This module depends on
 [pyyaml](https://pypi.org/project/PyYAML/) package.
 """
-from typing import Type, Any, overload
+from typing import Type, Any, overload, Optional
 
 import yaml
 
@@ -25,7 +25,9 @@ class YamlDeserializer(Deserializer[str]):
         return yaml.safe_load(data, **opts)
 
 
-def to_yaml(obj: Any, se: Type[Serializer[str]] = YamlSerializer, **opts: Any) -> str:
+def to_yaml(
+    obj: Any, cls: Optional[Any] = None, se: Type[Serializer[str]] = YamlSerializer, **opts: Any
+) -> str:
     """
     Serialize the object into YAML.
 
@@ -35,7 +37,7 @@ def to_yaml(obj: Any, se: Type[Serializer[str]] = YamlSerializer, **opts: Any) -
     If you want to use the other yaml package, you can subclass `YamlSerializer` and implement
     your own logic.
     """
-    return se.serialize(to_dict(obj, reuse_instances=False), **opts)
+    return se.serialize(to_dict(obj, c=cls, reuse_instances=False), **opts)
 
 
 @overload


Test patch:
diff --git a/tests/test_union.py b/tests/test_union.py
index 658e44d..d51b468 100644
--- a/tests/test_union.py
+++ b/tests/test_union.py
@@ -7,9 +7,19 @@ from uuid import UUID
 
 import pytest
 
-from serde import SerdeError, from_dict, from_tuple
-from serde import init as serde_init
-from serde import logger, serde, to_dict, to_tuple
+from serde import (
+    SerdeError,
+    from_dict,
+    from_tuple,
+    init as serde_init,
+    logger,
+    serde,
+    to_dict,
+    to_tuple,
+    InternalTagging,
+    AdjacentTagging,
+    Untagged,
+)
 from serde.compat import Literal
 from serde.json import from_json, to_json
 
@@ -572,7 +582,7 @@ def test_internal_tagging():
     with pytest.raises(Exception):
         assert from_dict(Foo, {"a": {"type": "Bar", "c": 10}})
 
-    with pytest.raises(SerdeError):
+    with pytest.raises(TypeError):
         # Tag is not specified in attribute
         @serde(tagging=InternalTagging())
         class Foo:
@@ -632,19 +642,19 @@ def test_adjacent_tagging():
     with pytest.raises(Exception):
         assert from_dict(Foo, {"a": {"type": "Bar", "content": {"c": 10}}})
 
-    with pytest.raises(SerdeError):
+    with pytest.raises(TypeError):
         # Tag is not specified in attribute
         @serde(tagging=AdjacentTagging(content="content"))
         class Foo:
             pass
 
-    with pytest.raises(SerdeError):
+    with pytest.raises(TypeError):
         # Content is not specified in attribute
         @serde(tagging=AdjacentTagging(tag="tag"))
         class Foo:
             pass
 
-    with pytest.raises(SerdeError):
+    with pytest.raises(TypeError):
         # Tag/Content is not specified in attribute
         @serde(tagging=AdjacentTagging())
         class Foo:
@@ -725,3 +735,31 @@ def test_newtype_and_untagged_union() -> None:
 
     assert isinstance(actual.buggy_field[0], Innerclass)
     assert isinstance(actual.buggy_field[1], str)
+
+
+def test_union_directly() -> None:
+    @dataclass
+    class Foo:
+        v: int
+
+    @dataclass
+    class Bar:
+        w: str
+
+    bar = Bar("bar")
+
+    # externally tagged
+    s = to_json(bar, cls=Union[Foo, Bar])
+    assert bar == from_json(Union[Foo, Bar], s)
+
+    # internally tagged
+    s = to_json(bar, cls=InternalTagging("type", Union[Foo, Bar]))
+    assert bar == from_json(InternalTagging("type", Union[Foo, Bar]), s)
+
+    # adjacently tagged
+    s = to_json(bar, cls=AdjacentTagging("type", "content", Union[Foo, Bar]))
+    assert bar == from_json(AdjacentTagging("type", "content", Union[Foo, Bar]), s)
+
+    # untagged tagged
+    s = to_json(bar, cls=Untagged(Union[Foo, Bar]))
+    assert bar == from_json(Untagged(Union[Foo, Bar]), s)