Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,46 @@ print(vars(public_user_info))
# {'full_name': 'John Cusack', 'profession': 'engineer'}
```

## Use of Deepcopy
By default, automapper performs a recursive deepcopy() on all attributes. This makes sure that changes in the attributes of the source
do not affect the target and vice-versa:

```python
from dataclasses import dataclass
from automapper import mapper

@dataclass
class Address:
street: str
number: int
zip_code: int
city: str

class PersonInfo:
def __init__(self, name: str, age: int, address: Address):
self.name = name
self.age = age
self.address = address

class PublicPersonInfo:
def __init__(self, name: str, address: Address):
self.name = name
self.address = address

address = Address(street="Main Street", number=1, zip_code=100001, city='Test City')
info = PersonInfo('John Doe', age=35, address=address)

public_info = mapper.to(PublicPersonInfo).map(info)
assert address is not public_info.address
```

To disable this behavior, you may pass `deepcopy=False` to either `mapper.map()` or to `mapper.add()`. If both are passed,
the argument of the `.map()` call has priority. E.g.

```python
public_info = mapper.to(PublicPersonInfo).map(info, deepcopy=False)
assert address is public_info.address
```

## Extensions
`py-automapper` has few predefined extensions for mapping support to classes for frameworks:
Expand Down
4 changes: 1 addition & 3 deletions automapper/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,4 @@ class MappingError(Exception):

class CircularReferenceError(Exception):
def __init__(self, *args: object) -> None:
super().__init__(
"Mapper does not support objects with circular references yet", *args
)
super().__init__("Mapper does not support objects with circular references yet", *args)
4 changes: 1 addition & 3 deletions automapper/extensions/default.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,7 @@ def __init_method_classifier__(target_cls: Type[T]) -> bool:
return (
hasattr(target_cls, "__init__")
and hasattr(getattr(target_cls, "__init__"), "__annotations__")
and isinstance(
getattr(getattr(target_cls, "__init__"), "__annotations__"), dict
)
and isinstance(getattr(getattr(target_cls, "__init__"), "__annotations__"), dict)
and getattr(getattr(target_cls, "__init__"), "__annotations__")
)

Expand Down
48 changes: 30 additions & 18 deletions automapper/mapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,26 +65,29 @@ def map(
*,
skip_none_values: bool = False,
fields_mapping: FieldsMap = None,
deepcopy: bool = True,
) -> T:
"""Produces output object mapped from source object and custom arguments

Parameters:
skip_none_values - do not map fields that has None value
fields_mapping - mapping for fields with different names
deepcopy - should we deepcopy all attributes? [default: True]
"""
return self.__mapper._map_common(
obj,
self.__target_cls,
set(),
skip_none_values=skip_none_values,
fields_mapping=fields_mapping,
deepcopy=deepcopy,
)


class Mapper:
def __init__(self) -> None:
"""Initializes internal containers"""
self._mappings: Dict[Type[S], Tuple[T, FieldsMap]] = {} # type: ignore [valid-type]
self._mappings: Dict[Type[S], Tuple[T, FieldsMap, bool]] = {} # type: ignore [valid-type]
self._class_specs: Dict[Type[T], SpecFunction[T]] = {} # type: ignore [valid-type]
self._classifier_specs: Dict[ # type: ignore [valid-type]
ClassifierFunction[T], SpecFunction[T]
Expand All @@ -102,9 +105,7 @@ def add_spec(self, classifier: Type[T], spec_func: SpecFunction[T]) -> None:
...

@overload
def add_spec(
self, classifier: ClassifierFunction[T], spec_func: SpecFunction[T]
) -> None:
def add_spec(self, classifier: ClassifierFunction[T], spec_func: SpecFunction[T]) -> None:
"""Add a spec function for all classes identified by classifier function.

Parameters:
Expand Down Expand Up @@ -141,6 +142,7 @@ def add(
target_cls: Type[T],
override: bool = False,
fields_mapping: FieldsMap = None,
deepcopy: bool = True,
) -> None:
"""Adds mapping between object of `source class` to an object of `target class`.

Expand All @@ -152,35 +154,37 @@ def add(
Target class to map to
override : bool, optional
Override existing `source class` mapping to use new `target class`
deepcopy : bool, optional
Should we deepcopy all attributes? [default: True]
"""
if source_cls in self._mappings and not override:
raise DuplicatedRegistrationError(
f"source_cls {source_cls} was already added for mapping"
)
self._mappings[source_cls] = (target_cls, fields_mapping)
self._mappings[source_cls] = (target_cls, fields_mapping, deepcopy)

def map(
self,
obj: object,
*,
skip_none_values: bool = False,
fields_mapping: FieldsMap = None,
deepcopy: bool = None,
) -> T: # type: ignore [type-var]
"""Produces output object mapped from source object and custom arguments"""
obj_type = type(obj)
if obj_type not in self._mappings:
raise MappingError(f"Missing mapping type for input type {obj_type}")
obj_type_preffix = f"{obj_type.__name__}."

target_cls, target_cls_field_mappings = self._mappings[obj_type]
target_cls, target_cls_field_mappings, target_deepcopy = self._mappings[obj_type]

common_fields_mapping = fields_mapping
if target_cls_field_mappings:
# transform mapping if it's from source class field
common_fields_mapping = {
target_obj_field: getattr(obj, source_field[len(obj_type_preffix) :])
if isinstance(source_field, str)
and source_field.startswith(obj_type_preffix)
if isinstance(source_field, str) and source_field.startswith(obj_type_preffix)
else source_field
for target_obj_field, source_field in target_cls_field_mappings.items()
}
Expand All @@ -190,12 +194,17 @@ def map(
**fields_mapping,
} # merge two dict into one, fields_mapping has priority

# If deepcopy is not explicitly given, we use target_deepcopy
if deepcopy is None:
deepcopy = target_deepcopy

return self._map_common(
obj,
target_cls,
set(),
skip_none_values=skip_none_values,
fields_mapping=common_fields_mapping,
deepcopy=deepcopy,
)

def _get_fields(self, target_cls: Type[T]) -> Iterable[str]:
Expand All @@ -208,9 +217,7 @@ def _get_fields(self, target_cls: Type[T]) -> Iterable[str]:
if classifier(target_cls):
return self._classifier_specs[classifier](target_cls)

raise MappingError(
f"No spec function is added for base class of {type(target_cls)}"
)
raise MappingError(f"No spec function is added for base class of {type(target_cls)}")

def _map_subobject(
self, obj: S, _visited_stack: Set[int], skip_none_values: bool = False
Expand All @@ -224,7 +231,7 @@ def _map_subobject(
raise CircularReferenceError()

if type(obj) in self._mappings:
target_cls, _ = self._mappings[type(obj)]
target_cls, _, _ = self._mappings[type(obj)]
result: Any = self._map_common(
obj, target_cls, _visited_stack, skip_none_values=skip_none_values
)
Expand All @@ -234,9 +241,7 @@ def _map_subobject(
if is_sequence(obj):
if isinstance(obj, dict):
result = {
k: self._map_subobject(
v, _visited_stack, skip_none_values=skip_none_values
)
k: self._map_subobject(v, _visited_stack, skip_none_values=skip_none_values)
for k, v in obj.items()
}
else:
Expand All @@ -262,12 +267,14 @@ def _map_common(
_visited_stack: Set[int],
skip_none_values: bool = False,
fields_mapping: FieldsMap = None,
deepcopy: bool = True,
) -> T:
"""Produces output object mapped from source object and custom arguments

Parameters:
skip_none_values - do not map fields that has None value
fields_mapping - fields mappings for fields with different names
deepcopy - Should we deepcopy all attributes? [default: True]
"""
obj_id = id(obj)

Expand All @@ -293,9 +300,14 @@ def _map_common(
value = obj[field_name] # type: ignore [index]

if value is not None:
mapped_values[field_name] = self._map_subobject(
value, _visited_stack, skip_none_values
)
if deepcopy:
mapped_values[field_name] = self._map_subobject(
value, _visited_stack, skip_none_values
)
else:
# if deepcopy is disabled, we can act as if value was a primitive type and
# avoid the ._map_subobject() call entirely.
mapped_values[field_name] = value
elif not skip_none_values:
mapped_values[field_name] = None

Expand Down
12 changes: 3 additions & 9 deletions tests/test_automapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,7 @@ def __init__(self, num: int, text: str, flag: bool) -> None:

@classmethod
def fields(cls) -> Iterable[str]:
return (
field for field in cls.__init__.__annotations__.keys() if field != "return"
)
return (field for field in cls.__init__.__annotations__.keys() if field != "return")


class AnotherClass:
Expand Down Expand Up @@ -68,16 +66,12 @@ def setUp(self):
def test_add_spec__adds_to_internal_collection(self):
self.mapper.add_spec(ParentClass, custom_spec_func)
assert ParentClass in self.mapper._class_specs
assert ["num", "text", "flag"] == self.mapper._class_specs[ParentClass](
ChildClass
)
assert ["num", "text", "flag"] == self.mapper._class_specs[ParentClass](ChildClass)

def test_add_spec__error_on_adding_same_class_spec(self):
self.mapper.add_spec(ParentClass, custom_spec_func)
with pytest.raises(DuplicatedRegistrationError):
self.mapper.add_spec(
ParentClass, lambda concrete_type: ["field1", "field2"]
)
self.mapper.add_spec(ParentClass, lambda concrete_type: ["field1", "field2"])

def test_add_spec__adds_to_internal_collection_for_classifier(self):
self.mapper.add_spec(classifier_func, spec_func)
Expand Down
74 changes: 51 additions & 23 deletions tests/test_automapper_dict_field.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from copy import deepcopy
from typing import Any, Dict
from unittest import TestCase

from automapper import mapper
from automapper import mapper, create_mapper


class Candy:
Expand All @@ -12,36 +12,64 @@ def __init__(self, name: str, brand: str):

class Shop:
def __init__(self, products: Dict[str, Any], annual_income: int):
self.products: Dict[str, Any] = deepcopy(products)
self.products: Dict[str, Any] = products
self.annual_income = annual_income


class ShopPublicInfo:
def __init__(self, products: Dict[str, Any]):
self.products: Dict[str, Any] = deepcopy(products)
self.products: Dict[str, Any] = products


def test_map__with_dict_field():
products = {
"magazines": ["Forbes", "Time", "The New Yorker"],
"candies": [
Candy("Reese's cups", "The Hershey Company"),
Candy("Snickers", "Mars, Incorporated"),
],
}
shop = Shop(products=products, annual_income=10000000)
class AutomapperTest(TestCase):
def setUp(self) -> None:
products = {
"magazines": ["Forbes", "Time", "The New Yorker"],
"candies": [
Candy("Reese's cups", "The Hershey Company"),
Candy("Snickers", "Mars, Incorporated"),
],
}
self.shop = Shop(products=products, annual_income=10000000)
self.mapper = create_mapper()

public_info = mapper.to(ShopPublicInfo).map(shop)
def test_map__with_dict_field(self):
public_info = mapper.to(ShopPublicInfo).map(self.shop)

assert public_info.products["magazines"] == shop.products["magazines"]
assert id(public_info.products["magazines"]) != id(shop.products["magazines"])
self.assertEqual(public_info.products["magazines"], self.shop.products["magazines"])
self.assertNotEqual(
id(public_info.products["magazines"]), id(self.shop.products["magazines"])
)

assert public_info.products["candies"] != shop.products["candies"]
assert public_info.products["candies"][0] != shop.products["candies"][0]
assert public_info.products["candies"][1] != shop.products["candies"][1]
self.assertNotEqual(public_info.products["candies"], self.shop.products["candies"])
self.assertNotEqual(public_info.products["candies"][0], self.shop.products["candies"][0])
self.assertNotEqual(public_info.products["candies"][1], self.shop.products["candies"][1])

assert public_info.products["candies"][0].name == "Reese's cups"
assert public_info.products["candies"][0].brand == "The Hershey Company"
self.assertEqual(public_info.products["candies"][0].name, "Reese's cups")
self.assertEqual(public_info.products["candies"][0].brand, "The Hershey Company")

assert public_info.products["candies"][1].name == "Snickers"
assert public_info.products["candies"][1].brand == "Mars, Incorporated"
self.assertEqual(public_info.products["candies"][1].name, "Snickers")
self.assertEqual(public_info.products["candies"][1].brand, "Mars, Incorporated")

def test_deepcopy_disabled(self):
public_info_deep = mapper.to(ShopPublicInfo).map(self.shop, deepcopy=False)
public_info = mapper.to(ShopPublicInfo).map(self.shop)

self.assertIsNot(public_info.products, self.shop.products)
self.assertEqual(public_info.products["magazines"], self.shop.products["magazines"])
self.assertNotEqual(public_info.products["magazines"], id(self.shop.products["magazines"]))

self.assertIs(public_info_deep.products, self.shop.products)
self.assertEqual(
id(public_info_deep.products["magazines"]), id(self.shop.products["magazines"])
)

def test_deepcopy_disabled_in_add(self):
self.mapper.add(Shop, ShopPublicInfo, deepcopy=False)
public_info = self.mapper.map(self.shop)

self.assertIs(public_info.products, self.shop.products)

# Manually enable deepcopy on .map()
public_info = self.mapper.map(self.shop, deepcopy=True)
self.assertIsNot(public_info.products, self.shop.products)
Loading