Repository: onicagroup/runway

Issue:
use pydantic to parse arguments for all `runway.cfngin.hooks.docker` hooks

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/onicagroup/runway/pull/1219?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) Report
> Merging [#1219](https://codecov.io/gh/onicagroup/runway/pull/1219?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) (8cfbfac) into [master](https://codecov.io/gh/onicagroup/runway/commit/62cf452c1fe6192ec519a5592d913aa6202d47b4?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) (62cf452) will **decrease** coverage by `0.11%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/onicagroup/runway/pull/1219/graphs/tree.svg?width=650&height=150&src=pr&token=Ku28I0RY80&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup)](https://codecov.io/gh/onicagroup/runway/pull/1219?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup)

```diff
@@            Coverage Diff             @@
##           master    #1219      +/-   ##
==========================================
- Coverage   89.52%   89.41%   -0.12%     
==========================================
  Files         247      247              
  Lines       15312    15218      -94     
==========================================
- Hits        13708    13607     -101     
- Misses       1604     1611       +7     
```


| [Impacted Files](https://codecov.io/gh/onicagroup/runway/pull/1219?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) | Coverage Δ | |
|---|---|---|
| [runway/cfngin/hooks/docker/\_login.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9kb2NrZXIvX2xvZ2luLnB5) | `100.00% <100.00%> (ø)` | |
| [runway/cfngin/hooks/docker/data\_models.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9kb2NrZXIvZGF0YV9tb2RlbHMucHk=) | `100.00% <100.00%> (ø)` | |
| [runway/cfngin/hooks/docker/image/\_build.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9kb2NrZXIvaW1hZ2UvX2J1aWxkLnB5) | `100.00% <100.00%> (ø)` | |
| [runway/cfngin/hooks/docker/image/\_push.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9kb2NrZXIvaW1hZ2UvX3B1c2gucHk=) | `100.00% <100.00%> (ø)` | |
| [runway/cfngin/hooks/docker/image/\_remove.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9kb2NrZXIvaW1hZ2UvX3JlbW92ZS5weQ==) | `100.00% <100.00%> (ø)` | |
| [runway/cfngin/hooks/aws\_lambda.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9hd3NfbGFtYmRhLnB5) | `84.10% <0.00%> (-1.66%)` | :arrow_down: |
| [runway/env\_mgr/kbenv.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Vudl9tZ3Iva2JlbnYucHk=) | `78.76% <0.00%> (-0.69%)` | :arrow_down: |
| [runway/env\_mgr/tfenv.py](https://codecov.io/gh/onicagroup/runway/pull/1219/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Vudl9tZ3IvdGZlbnYucHk=) | `96.46% <0.00%> (-0.51%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/onicagroup/runway/pull/1219?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/onicagroup/runway/pull/1219?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup). Last update [62cf452...8cfbfac](https://codecov.io/gh/onicagroup/runway/pull/1219?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup).


Patch:
diff --git a/Makefile b/Makefile
index da3347c6..b8ad4095 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: build  clean help install lint list release test version
+.PHONY: build clean docs help install lint list release test version
 
 SHELL := /bin/bash
 
diff --git a/runway/cfngin/hooks/docker/_login.py b/runway/cfngin/hooks/docker/_login.py
index 6bc41574..3af9ea72 100644
--- a/runway/cfngin/hooks/docker/_login.py
+++ b/runway/cfngin/hooks/docker/_login.py
@@ -1,67 +1,14 @@
-"""Docker login hook.
-
-Replicates the functionality of the ``docker login`` CLI command.
-
-This hook does not modify the Docker config file.
-The credentials/authenticated session is stored in memory and is deleted after
-processing a given CFNgin config file.
-
-This hook can be executed as a pre or post hook.
-The authenticated session carries over from pre to post and to each subsequent
-built-in Docker hook.
-
-.. rubric:: Hook Path
-
-``runway.cfngin.hooks.docker.login``
-
-.. rubric:: Args
-
-dockercfg_path (Optional[str])
-    Use a custom path for the Docker config file (``$HOME/.docker/config.json``
-    if present, otherwise ``$HOME/.dockercfg``).
-ecr (Optional[Union[bool, Dict[str, Optional[str]]]])
-    Information describing an ECR registry. This is used to construct the registry URL.
-    If providing a value for this field, do not provide a value for ``registry``.
-
-    If using a private registry, the value can simply be ``true``.
-    If using a public registry, more information is required.
-
-    account_id (Optional[str])
-        AWS account ID that owns the registry being logged into. If not provided,
-        it will be acquired automatically if needed.
-    alias (Optional[str])
-        If it is a public repository, provide the alias.
-    aws_region (Optional[str])
-        AWS region where the registry is located. If not provided, it will be acquired
-        automatically if needed.
-
-email (Optional[str])
-    The email for the registry account.
-password (str)
-    The plaintext password.
-registry (Optional[str])
-    URL to the registry (e.g. ``https://index.docker.io/v1/``).
-
-    If providing a value for this field, do not provide a value for ``ecr``.
-username (str)
-    The registry username. Defaults to ``AWS`` if supplying ``ecr``.
-
-.. rubric:: Example
-.. code-block:: yaml
-
-    pre_deploy:
-      - path: runway.cfngin.hooks.docker.login
-        args:
-          ecr: true
-          password: ${ecr login-password}
-
-"""
+"""Docker login hook."""
+# pylint: disable=no-self-argument,no-self-use
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Dict, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Dict, Optional
 
-from .data_models import BaseModel, ElasticContainerRegistry
+from pydantic import Field, validator
+
+from ....utils import BaseModel
+from .data_models import ElasticContainerRegistry
 from .hook_data import DockerHookData
 
 if TYPE_CHECKING:
@@ -82,61 +29,45 @@ class LoginArgs(BaseModel):
 
     """
 
-    def __init__(
-        self,
-        *,
-        context: Optional[CfnginContext] = None,
-        dockercfg_path: Optional[str] = None,
-        ecr: Optional[Union[bool, Dict[str, Optional[str]]]] = None,
-        email: Optional[str] = None,
-        password: str,
-        registry: Optional[str] = None,
-        username: Optional[str] = None,
-        **_: Any,
-    ) -> None:
-        """Instantiate class.
-
-        Args:
-            context: CFNgin context object.
-            dockercfg_path: Use a custom path for the Docker config file
-                (``$HOME/.docker/config.json`` if present, otherwise ``$HOME/.dockercfg``).
-            ecr: Information describing an ECR registry. This is used to construct
-                the registry URL. If providing a value for this field, do not provide
-                a value for ``registry``.
-            email: The email for the registry account.
-            password: The plaintext password for the registry account.
-            registry: URL to the registry (e.g. ``https://index.docker.io/v1/``).
-            username: The registry username.
-                Defaults to ``AWS`` if supplying ``ecr``.
-
-        """
-        super().__init__(context=context)
-        self.dockercfg_path = self._validate_str(dockercfg_path, optional=True)
-        self.email = self._validate_str(email, optional=True)
-        self.password = cast(str, self._validate_str(password, required=True))
-        self.registry = self.determine_registry(
-            context=context, ecr=ecr, registry=registry
-        )
-        if not username:
-            self.username = cast(
-                str, ("AWS" if ecr else self._validate_str(username, required=True))
+    _ctx: Optional[CfnginContext] = Field(default=None, alias="context", exclude=True)
+
+    dockercfg_path: Optional[str] = None
+    """Path to a non-default Docker config file."""
+
+    ecr: Optional[ElasticContainerRegistry] = Field(default=None, exclude=True)
+    """Information describing an ECR registry."""
+
+    email: Optional[str] = None
+    """The email for the registry account."""
+
+    password: str
+    """The plaintext password for the registry account."""
+
+    registry: Optional[str] = None
+    """URI of the registry to login to."""
+
+    username: str = "AWS"
+    """The registry username."""
+
+    @validator("ecr", pre=True, allow_reuse=True)
+    def _set_ecr(cls, v: Any, values: Dict[str, Any]) -> Any:
+        """Set the value of ``ecr``."""
+        if v and isinstance(v, dict):
+            return ElasticContainerRegistry.parse_obj(
+                {"context": values.get("context"), **v}
             )
-        else:
-            self.username = cast(str, self._validate_str(username, required=True))
-
-    @staticmethod
-    def determine_registry(
-        context: Optional[CfnginContext] = None,
-        ecr: Optional[Union[bool, Dict[str, Optional[str]]]] = None,
-        registry: Optional[str] = None,
-    ) -> Optional[str]:
-        """Determine repository URI."""
-        if registry:
-            return registry
+        return v
+
+    @validator("registry", pre=True, always=True, allow_reuse=True)
+    def _set_registry(cls, v: Any, values: Dict[str, Any]) -> Any:
+        """Set the value of ``registry``."""
+        if v:
+            return v
+
+        ecr: Optional[ElasticContainerRegistry] = values.get("ecr")
         if ecr:
-            return ElasticContainerRegistry.parse_obj(
-                ecr if isinstance(ecr, dict) else {}, context=context  # type: ignore
-            ).fqn
+            return ecr.fqn
+
         return None
 
 
@@ -148,8 +79,7 @@ def login(*, context: CfnginContext, **kwargs: Any) -> DockerHookData:
     kwargs are parsed by :class:`~runway.cfngin.hooks.docker.LoginArgs`.
 
     """
-    kwargs.pop("provider", None)
-    args = LoginArgs.parse_obj(kwargs, context=context)
+    args = LoginArgs.parse_obj({"context": context, **kwargs})
     docker_hook_data = DockerHookData.from_cfngin_context(context)
     docker_hook_data.client.login(**args.dict())
     LOGGER.info("logged into %s", args.registry)
diff --git a/runway/cfngin/hooks/docker/data_models.py b/runway/cfngin/hooks/docker/data_models.py
index 7d97cb56..b1e0b063 100644
--- a/runway/cfngin/hooks/docker/data_models.py
+++ b/runway/cfngin/hooks/docker/data_models.py
@@ -4,272 +4,26 @@ These are makeshift data models for use until Runway v2 is realeased and pydanti
 can be used.
 
 """
+# pylint: disable=no-self-argument,no-self-use
 from __future__ import annotations
 
-from pathlib import Path
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    ClassVar,
-    Dict,
-    Iterable,
-    List,
-    NoReturn,
-    Optional,
-    Tuple,
-    Type,
-    TypeVar,
-    Union,
-    cast,
-    overload,
-)
+from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, cast
 
-from typing_extensions import Literal
+from docker.models.images import Image
+from pydantic import Field, root_validator
 
 from ....core.providers.aws import AccountDetails
-from ....utils import MutableMap
+from ....utils import BaseModel, MutableMap
 
 if TYPE_CHECKING:
-    from docker.models.images import Image
-
     from ....context import CfnginContext
 
-    Model = TypeVar("Model", bound="BaseModel")
-
 
 ECR_REPO_FQN_TEMPLATE = (
     "{aws_account_id}.dkr.ecr.{aws_region}.amazonaws.com/{repo_name}"
 )
 
 
-class BaseModel:
-    """Base model."""
-
-    def __init__(
-        self, *, context: Optional[CfnginContext] = None, **_kwargs: Any
-    ) -> None:
-        """Instantiate class."""
-        self._ctx = context
-
-    def dict(self) -> Dict[str, Any]:
-        """Return object as a dict."""
-        return {k: v for k, v in self.__iter__() if not k.startswith("_")}
-
-    def find(self, query: str, default: Any = None, **kwargs: Any) -> Any:
-        """Find a value in the object."""
-        split_query = query.split(".")
-
-        if len(split_query) == 1:
-            return self.get(split_query[0], default, **kwargs)
-
-        nested_value = self.get(split_query[0])
-
-        if not nested_value:
-            if self.get(query):
-                return self.get(query)
-            return default
-
-        try:
-            nested_value = nested_value.find(
-                query=".".join(split_query[1:]), default=default, **kwargs
-            )
-            return nested_value
-        except (AttributeError, KeyError):
-            return default
-
-    def get(self, name: str, default: Any = None) -> Any:
-        """Get a value or return default if it is not found.
-
-        Attr:
-            name: The value to look for.
-            default: Returned if no other value is found.
-
-        """
-        return getattr(self, name, default)
-
-    @staticmethod
-    def _validate_bool(value: Any) -> bool:
-        """Validate a bool type attribute."""
-        if isinstance(value, bool):
-            return value
-        return bool(value)
-
-    @overload
-    @classmethod
-    def _validate_dict(cls, value: Dict[str, Any]) -> Dict[str, Any]:
-        ...
-
-    @overload
-    @classmethod
-    def _validate_dict(cls, value: Optional[Dict[str, Any]]) -> Dict[str, Any]:
-        ...
-
-    @overload
-    @classmethod
-    def _validate_dict(
-        cls, value: Optional[Dict[str, Any]], optional: Literal[True]
-    ) -> Optional[Dict[str, Any]]:
-        ...
-
-    @overload
-    @classmethod
-    def _validate_dict(
-        cls, value: Dict[str, Any], optional: bool = ..., required: bool = ...
-    ) -> Dict[str, str]:
-        ...
-
-    @overload
-    @classmethod
-    def _validate_dict(
-        cls, value: None, optional: bool = ..., required: Literal[True] = ...
-    ) -> NoReturn:
-        ...
-
-    @overload
-    @classmethod
-    def _validate_dict(
-        cls, value: Literal[None], optional: Literal[True], required: Literal[False]
-    ) -> None:
-        ...
-
-    @classmethod
-    def _validate_dict(
-        cls,
-        value: Optional[Union[Dict[str, Any], Any]],
-        optional: bool = False,
-        required: bool = False,
-    ) -> Optional[Union[Dict[str, Any], NoReturn]]:
-        """Validate a Dict type attribute."""
-        if not value:
-            if required:
-                raise ValueError("Dict can't be empty or NoneType")
-            if not isinstance(value, dict):
-                if optional:
-                    return None
-                return {}
-        if isinstance(value, dict):
-            return value
-        return cls._validate_dict(dict(value), optional=optional, required=required)
-
-    @classmethod
-    def _validate_int(
-        cls, value: Any, optional: bool = False, required: bool = False
-    ) -> Optional[int]:
-        """Validate int type attribute."""
-        if not value and value != 0:
-            if required:
-                raise ValueError("int can't be NoneType")
-            if optional:
-                return None
-        if isinstance(value, int):
-            return value
-        return cls._validate_int(int(value or 0), optional=optional, required=required)
-
-    @classmethod
-    def _validate_list_str(
-        cls,
-        value: Union[List[str], Any],
-        optional: bool = False,
-        required: bool = False,
-    ) -> Optional[List[str]]:
-        """Validate a List[str] type attribute."""
-        if not value:
-            if required:
-                raise ValueError("List can't be empty or NoneType")
-            if not isinstance(value, list):
-                if optional:
-                    return None
-                return []
-        if isinstance(value, list):
-            if all(isinstance(i, str) for i in value):
-                return value
-            for i in value:
-                if not isinstance(i, str):
-                    raise TypeError(f"expected List[str] not List[{type(i).__name__}]")
-        return cls._validate_list_str(  # type: ignore
-            list(value), optional=optional, required=required
-        )
-
-    @staticmethod
-    def _validate_path(value: Any, must_exist: bool = False) -> Path:
-        """Validate a Path type attribute.
-
-        Args:
-            value: A Path object or string that can be converted into a Path object.
-            must_exist: Raise an error if the path provided does not exist.
-
-        """
-        if isinstance(value, str):
-            value = Path(value)
-        if not isinstance(value, Path):
-            raise TypeError(f"expected Union[Path, str] not {type(value).__name__}")
-        if must_exist and not value.exists():
-            raise ValueError(f"provided path does not exist: {value.resolve()}")
-        return value
-
-    @classmethod
-    def _validate_str(
-        cls, value: Any, optional: bool = False, required: bool = False
-    ) -> Optional[str]:
-        """Validate str type attribute."""
-        if not value:
-            if required:
-                raise ValueError("value can't be empty or NoneType")
-            if not isinstance(value, str) and optional:
-                return None
-        if isinstance(value, str):
-            return value
-        if isinstance(value, (dict, list, set, tuple)):
-            raise TypeError(f"value can't be {type(value).__name__}")  # type: ignore
-        return cls._validate_str(str(value), optional=optional, required=required)
-
-    @classmethod
-    def parse_obj(
-        cls: Type[Model], obj: Any, context: Optional[CfnginContext] = None
-    ) -> Model:
-        """Parse object."""
-        if not isinstance(obj, dict):
-            try:
-                obj = dict(obj)
-            except (TypeError, ValueError):
-                raise TypeError(
-                    f"{cls.__name__} expected dict not {obj.__class__.__name__}"
-                ) from None
-        return cls(context=context, **obj)
-
-    def __eq__(self, other: Any) -> bool:
-        """Evaluate equal comparison operator."""
-        if isinstance(other, self.__class__):
-            return self.dict() == other.dict()
-        return self.dict() == other
-
-    def __getitem__(self, name: str) -> bool:
-        """Implement evaluation of self[name].
-
-        Args:
-            name: The value to look for.
-
-        Raises:
-            KeyError: Object does not contain a field of the name provided.
-
-        """
-        return getattr(self, name)
-
-    def __iter__(self) -> Iterable[Tuple[str, Any]]:
-        """Iterate object."""
-        yield from self.__dict__.items()
-
-    def __setitem__(self, name: str, value: Any) -> None:
-        """Implement assignment to self[key].
-
-        Args:
-            name: Attribute name to associate with a value.
-            value: Value of a key/attribute.
-
-        """
-        setattr(self, name, value)
-
-
 class ElasticContainerRegistry(BaseModel):
     """AWS Elastic Container Registry.
 
@@ -284,37 +38,20 @@ class ElasticContainerRegistry(BaseModel):
     PUBLIC_URI_TEMPLATE: ClassVar[str] = "public.ecr.aws/{registry_alias}/"
     URI_TEMPLATE: ClassVar[str] = "{aws_account_id}.dkr.ecr.{aws_region}.amazonaws.com/"
 
-    account_id: Optional[str]
-    alias: Optional[str]
-    region: Optional[str]
-    public: bool
-
-    def __init__(
-        self,
-        *,
-        account_id: Optional[str] = None,
-        alias: Optional[str] = None,
-        aws_region: Optional[str] = None,
-        context: Optional[CfnginContext] = None,
-        **kwargs: Any,
-    ) -> None:
-        """Instantiate class."""
-        super().__init__(context=context, **kwargs)
-        self.account_id = self._validate_str(account_id, optional=True)
-        self.alias = self._validate_str(alias, optional=True)
-        self.region = self._validate_str(aws_region, optional=True)
-        self.public = bool(self.alias)
-
-        if not self.public:
-            if not self._ctx and not (self.account_id or self.region):
-                raise ValueError("context is required to resolve values")
-            if not self.region:
-                self.region = self._validate_str(
-                    self._ctx.env.aws_region if self._ctx else "us-east-1",
-                    required=True,
-                )
-            if not self.account_id:
-                self.account_id = AccountDetails(cast("CfnginContext", self._ctx)).id
+    _ctx: Optional[CfnginContext] = Field(default=None, alias="context", exclude=True)
+    """CFNgin context."""
+
+    account_id: Optional[str] = None
+    """AWS account ID that owns the registry being logged into."""
+
+    alias: Optional[str] = None
+    """If it is a public repository, the alias of the repository."""
+
+    public: bool = True
+    """Whether the repository is public."""
+
+    region: Optional[str] = Field(default=None, alias="aws_region")
+    """AWS region where the registry is located."""
 
     @property
     def fqn(self) -> str:
@@ -325,29 +62,44 @@ class ElasticContainerRegistry(BaseModel):
             aws_account_id=self.account_id, aws_region=self.region
         )
 
+    @root_validator(allow_reuse=True, pre=True)
+    def _set_defaults(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+        """Set default values based on other values."""
+        values.setdefault("public", bool(values.get("alias")))
+
+        if not values["public"]:
+            account_id = values.get("account_id")
+            ctx: Optional[CfnginContext] = values.get("context")
+            region = values.get("aws_region")
+            if not ctx and not (account_id or region):
+                raise ValueError("context is required to resolve values")
+            if ctx:
+                if not account_id:
+                    values["account_id"] = AccountDetails(ctx).id
+                if not region:
+                    values["aws_region"] = ctx.env.aws_region or "us-east-1"
+
+        return values
+
 
 class DockerImage(BaseModel):
     """Wrapper for :class:`docker.models.images.Image`."""
 
-    _repo: Optional[str]
     image: Image
+    _repo: Optional[str] = None
 
-    def __init__(self, *, image: Image, **kwargs: Any) -> None:
-        """Instantiate class."""
-        super().__init__(**kwargs)
-        self._repo = None
-        self.image = image
+    class Config:
+        """Model configuration."""
+
+        arbitrary_types_allowed = True
+        fields = {"_repo": {"exclude": True}}
+        underscore_attrs_are_private = True
 
     @property
     def id(self) -> str:
         """ID of the image."""
         return self.image.id
 
-    @id.setter
-    def id(self, value: str) -> None:
-        """Set the ID of the image."""
-        self.image.id = value  # type: ignore
-
     @property
     def repo(self) -> str:
         """Repository URI."""
@@ -355,21 +107,11 @@ class DockerImage(BaseModel):
             self._repo = self.image.attrs["RepoTags"][0].rsplit(":", 1)[0]
         return cast(str, self._repo)
 
-    @repo.setter
-    def repo(self, value: str) -> None:
-        """Set repository URI value."""
-        self._repo = value
-
     @property
     def short_id(self) -> str:
         """ID of the image truncated to 10 characters plus the ``sha256:`` prefix."""
         return self.image.short_id
 
-    @short_id.setter
-    def short_id(self, value: str) -> None:
-        """Set the ID of the image truncated to 10 characters plus the ``sha256:`` prefix."""
-        self.image.short_id = value  # type: ignore
-
     @property
     def tags(self) -> List[str]:
         """List of image tags."""
@@ -383,46 +125,13 @@ class DockerImage(BaseModel):
 
 
 class ElasticContainerRegistryRepository(BaseModel):
-    """AWS Elastic Container Registry (ECR) Repository.
-
-    Attributes:
-        name: The name of the repository.
-        registry: Information about an ECR registry.
+    """AWS Elastic Container Registry (ECR) Repository."""
 
-    """
+    name: str = Field(..., alias="repo_name")
+    """The name of the repository."""
 
-    name: str
     registry: ElasticContainerRegistry
-
-    def __init__(
-        self,
-        *,
-        account_id: Optional[str] = None,
-        aws_region: Optional[str] = None,
-        context: Optional[CfnginContext] = None,
-        registry_alias: Optional[str] = None,
-        repo_name: str,
-        **kwargs: Any,
-    ) -> None:
-        """Instantiace class.
-
-        Args:
-            account_id: AWS account ID.
-            aws_region: AWS region.
-            context: CFNgin context object.
-            registry_alias: Alias of a public ECR registry.
-            repo_name: Name of the ECR repository.
-
-        """
-        super().__init__(context=context, **kwargs)
-        self.name = cast(str, self._validate_str(repo_name, required=True))
-
-        self.registry = ElasticContainerRegistry(
-            account_id=account_id,
-            alias=registry_alias,
-            aws_region=aws_region,
-            context=context,
-        )
+    """Information about an ECR registry."""
 
     @property
     def fqn(self) -> str:
diff --git a/runway/cfngin/hooks/docker/image/_build.py b/runway/cfngin/hooks/docker/image/_build.py
index 9ea00da8..7c55f8d0 100644
--- a/runway/cfngin/hooks/docker/image/_build.py
+++ b/runway/cfngin/hooks/docker/image/_build.py
@@ -3,6 +3,7 @@
 Replicates the functionality of the ``docker image build`` CLI command.
 
 """
+# pylint: disable=no-self-argument,no-self-use
 from __future__ import annotations
 
 import logging
@@ -20,8 +21,14 @@ from typing import (
 )
 
 from docker.models.images import Image
+from pydantic import DirectoryPath, Field, validator
 
-from ..data_models import BaseModel, DockerImage, ElasticContainerRegistryRepository
+from .....utils import BaseModel
+from ..data_models import (
+    DockerImage,
+    ElasticContainerRegistry,
+    ElasticContainerRegistryRepository,
+)
 from ..hook_data import DockerHookData
 
 if TYPE_CHECKING:
@@ -31,219 +38,152 @@ LOGGER = logging.getLogger(__name__.replace("._", "."))
 
 
 class DockerImageBuildApiOptions(BaseModel):
-    """Options for controlling Docker.
-
-    Attributes:
-        buildargs: Dict of build-time variables that will be passed to Docker.
-        custom_context: Whether to use custom context when providing a file object.
-        extra_hosts: Extra hosts to add to `/etc/hosts` in the build containers.
-            Defined as a mapping of hostmane to IP address.
-        forcerm: Always remove intermediate containers, even after unsuccessful builds.
-        isolation: Isolation technology used during build.
-        network_mode: Network mode for the run commands during build.
-        nocache: Whether to use cache.
-        platform: Set platform if server is multi-platform capable.
-            Uses format ``os[/arch[/variant]]``.
-        pull: Whether to download any updates to the FROM image in the Dockerfile.
-        rm: Whether to remove intermediate containers.
-        squash: Whether to squash the resulting image layers into a single layer.
-        tag: Optional name and tag to apply to the base image when it is built.
-        target: Name of the build-stage to build in a multi-stage Dockerfile.
-        timeout: HTTP timeout.
-        use_config_proxy: If ``True`` and if the Docker client configuration file
-            (``~/.docker/config.json`` by default) contains a proxy configuration,
-            the corresponding environment variables will be set in the container
-            being built.
+    """Options for controlling Docker."""
+
+    buildargs: Dict[str, Any] = {}
+    """Dict of build-time variables that will be passed to Docker."""
+
+    custom_context: bool = False
+    """Whether to use custom context when providing a file object."""
+
+    extra_hosts: Dict[str, str] = {}
+    """Extra hosts to add to `/etc/hosts` in the build containers.
+    Defined as a mapping of hostmane to IP address.
+
+    """
+
+    forcerm: bool = False
+    """Always remove intermediate containers, even after unsuccessful builds."""
+
+    isolation: Optional[str] = None
+    """Isolation technology used during build."""
+
+    network_mode: Optional[str] = None
+    """Network mode for the run commands during build."""
+
+    nocache: bool = False
+    """Whether to use cache."""
+
+    platform: Optional[str] = None
+    """Set platform if server is multi-platform capable.
+    Uses format ``os[/arch[/variant]]``.
 
     """
 
-    buildargs: Dict[str, Any]
-    custom_context: bool
-    extra_hosts: Optional[Dict[str, str]]
-    forcerm: bool
-    isolation: Optional[str]
-    network_mode: Optional[str]
-    nocache: bool
-    platform: Optional[str]
-    pull: bool
-    rm: bool
-    squash: bool
-    tag: Optional[str]
-    target: Optional[str]
-    timout: Optional[int]
-    use_config_proxy: bool
-
-    def __init__(
-        self,
-        *,
-        buildargs: Optional[Dict[str, Any]] = None,
-        custom_context: bool = False,
-        extra_hosts: Optional[Dict[str, Any]] = None,
-        forcerm: bool = False,
-        isolation: Optional[str] = None,
-        network_mode: Optional[str] = None,
-        nocache: bool = False,
-        platform: Optional[str] = None,
-        pull: bool = False,
-        rm: bool = True,  # pylint: disable=invalid-name
-        squash: bool = False,
-        tag: Optional[str] = None,
-        target: Optional[str] = None,
-        timeout: Optional[int] = None,
-        use_config_proxy: bool = False,
-        **kwargs: Any,
-    ) -> None:
-        """Instantiate class.
-
-        Args:
-            buildargs: Dict of build-time variables that will be passed to Docker.
-            custom_context: Whether to use custom context when providing a file object.
-            extra_hosts: Extra hosts to add to `/etc/hosts` in the build containers.
-                Defined as a mapping of hostmane to IP address.
-            forcerm: Always remove intermediate containers, even after unsuccessful builds.
-            isolation: Isolation technology used during build.
-            network_mode: Network mode for the run commands during build.
-            nocache: Whether to use cache.
-            platform: Set platform if server is multi-platform capable.
-                Uses format ``os[/arch[/variant]]``.
-            pull: Whether to download any updates to the FROM image in the Dockerfile.
-            rm: Whether to remove intermediate containers.
-            squash: Whether to squash the resulting image layers into a single layer.
-            tag: Optional name and tag to apply to the base image when it is built.
-            target: Name of the build-stage to build in a multi-stage Dockerfile.
-            timeout: HTTP timeout.
-            use_config_proxy: If ``True`` and if the Docker client configuration file
-                (``~/.docker/config.json`` by default) contains a proxy configuration,
-                the corresponding environment variables will be set in the container
-                being built.
-
-        """
-        super().__init__(**kwargs)
-        self.buildargs = self._validate_dict(buildargs)
-        self.custom_context = self._validate_bool(custom_context)
-        self.extra_hosts = self._validate_dict(extra_hosts, optional=True)
-        self.forcerm = self._validate_bool(forcerm)
-        self.isolation = self._validate_str(isolation, optional=True)
-        self.network_mode = self._validate_str(network_mode, optional=True)
-        self.nocache = self._validate_bool(nocache)
-        self.platform = self._validate_str(platform, optional=True)
-        self.pull = self._validate_bool(pull)
-        self.rm = self._validate_bool(rm)  # pylint: disable=invalid-name
-        self.squash = self._validate_bool(squash)
-        self.tag = self._validate_str(tag, optional=True)
-        self.target = self._validate_str(target, optional=True)
-        self.timeout = self._validate_int(timeout, optional=True)
-        self.use_config_proxy = self._validate_bool(use_config_proxy)
+    pull: bool = False
+    """Whether to download any updates to the FROM image in the Dockerfile."""
+
+    rm: bool = True
+    """Whether to remove intermediate containers."""
+
+    squash: bool = False
+    """Whether to squash the resulting image layers into a single layer."""
+
+    tag: Optional[str] = None
+    """Optional name and tag to apply to the base image when it is built."""
+
+    target: Optional[str] = None
+    """Name of the build-stage to build in a multi-stage Dockerfile."""
+
+    timeout: Optional[int] = None
+    """HTTP timeout."""
+
+    use_config_proxy: bool = False
+    """If ``True`` and if the Docker client configuration file
+    (``~/.docker/config.json`` by default) contains a proxy configuration,
+    the corresponding environment variables will be set in the container
+    being built.
+
+    """
 
 
 class ImageBuildArgs(BaseModel):
-    """Args passed to image.build.
+    """Args passed to image.build."""
+
+    _ctx: Optional[CfnginContext] = Field(default=None, alias="context", export=False)
 
-    Attributes:
-        docker: Options for ``docker image build``.
-        dockerfile: Path within the build context to the Dockerfile.
-        ecr_repo: AWS Elastic Container Registry repository information.
-            Providing this will automatically construct the repo URI.
-            If provided, do not provide ``repo``.
+    ecr_repo: Optional[ElasticContainerRegistryRepository] = None  # depends on _ctx
+    """AWS Elastic Container Registry repository information.
+    Providing this will automatically construct the repo URI.
+    If provided, do not provide ``repo``.
 
-            If using a private registry, only ``repo_name`` is required.
-            If using a public registry, ``repo_name`` and ``registry_alias``.
-        path: Path to the directory containing the Dockerfile.
-        repo: URI of a non Docker Hub repository where the image will be stored.
-        tags: List of tags to apply to the image.
+    If using a private registry, only ``repo_name`` is required.
+    If using a public registry, ``repo_name`` and ``registry_alias``.
 
     """
 
-    docker: DockerImageBuildApiOptions
-    dockerfile: str
-    path: Path
-    repo: Optional[str]
-    tags: List[str]
-
-    def __init__(
-        self,
-        *,
-        context: Optional[CfnginContext] = None,
-        docker: Optional[Dict[str, Any]] = None,
-        dockerfile: str = "./Dockerfile",
-        ecr_repo: Optional[Dict[str, Optional[str]]] = None,
-        path: Optional[Union[Path, str]] = None,
-        repo: Optional[str] = None,
-        tags: Optional[List[str]] = None,
-        **_: Any,
-    ) -> None:
-        """Instantiate class.
-
-        Args:
-            context: CFNgin context object.
-            docker: Options for ``docker image build``.
-            dockerfile: Path within the build context to the Dockerfile.
-            ecr_repo: AWS Elastic Container Registry repository information.
-                Providing this will automatically create the repo URI.
-                If provided, do not provide ``repo``.
-            path: Path to the directory containing the Dockerfile.
-            repo: URI of a non Docker Hub repository where the image will be stored.
-                If providing one of the other repo values, leave this value empty.
-            tags: List of tags to apply to the image. If not provided, ``["latest"]``
-                will be used.
-
-        """
-        super().__init__(context=context)
-        docker = docker or {}
-        self.path = self._validate_path(path or Path.cwd(), must_exist=True)
-        self.dockerfile = self._validate_dockerfile(self.path, dockerfile)
-        self.repo = self.determine_repo(
-            context=context,
-            ecr_repo=self._validate_dict(ecr_repo, optional=True),
-            repo=self._validate_str(repo, optional=True),
-        )
-        self.tags = cast(
-            List[str], self._validate_list_str(tags or ["latest"], required=True)
-        )
-
-        if self.repo:
-            docker.setdefault("tag", self.repo)
-        self.docker = DockerImageBuildApiOptions.parse_obj(docker)
-
-    @classmethod
-    def _validate_dockerfile(cls, path: Path, dockerfile: str) -> str:
-        """Validate Dockerfile."""
-        if path.is_file():
-            if path.name.endswith("Dockerfile"):
-                raise ValueError(
-                    cls.__name__ + ".path should not reference the Dockerfile directly"
-                    " but the directory containing the Dockerfile"
-                )
-            return dockerfile
-        fq_dockerfile = path / dockerfile
-        if not fq_dockerfile.is_file():
-            raise ValueError(
-                f"Dockerfile does not exist at path provided: {fq_dockerfile}"
-            )
-        return dockerfile
-
-    @staticmethod
-    def determine_repo(
-        context: Optional[CfnginContext] = None,
-        ecr_repo: Optional[Dict[str, Optional[str]]] = None,
-        repo: Optional[str] = None,
-    ) -> Optional[str]:
-        """Determine repo URI.
-
-        Args:
-            context: CFNgin context.
-            ecr_repo: AWS Elastic Container Registry options.
-            repo: URI of a non Docker Hub repository.
-
-        """
+    path: DirectoryPath = Path.cwd()
+    """Path to the directory containing the Dockerfile."""
+
+    dockerfile: str = "Dockerfile"  # depends on path for validation
+    """Path within the build context to the Dockerfile."""
+
+    repo: Optional[str] = None  # depends on ecr_repo
+    """URI of a non Docker Hub repository where the image will be stored."""
+
+    docker: DockerImageBuildApiOptions = DockerImageBuildApiOptions()  # depends on repo
+    """Options for ``docker image build``."""
+
+    tags: List[str] = ["latest"]
+    """List of tags to apply to the image."""
+
+    @validator("docker", pre=True, always=True, allow_reuse=True)
+    def _set_docker(
+        cls,
+        v: Union[Dict[str, Any], DockerImageBuildApiOptions, Any],
+        values: Dict[str, Any],
+    ) -> Any:
+        """Set the value of ``docker``."""
+        repo = values["repo"]
         if repo:
-            return repo
-        if ecr_repo:
+            if isinstance(v, dict):
+                v.setdefault("tag", repo)
+            elif isinstance(v, DockerImageBuildApiOptions) and not v.tag:
+                v.tag = repo
+        return v
+
+    @validator("ecr_repo", pre=True, allow_reuse=True)
+    def _set_ecr_repo(cls, v: Any, values: Dict[str, Any]) -> Any:
+        """Set the value of ``ecr_repo``."""
+        if v and isinstance(v, dict):
             return ElasticContainerRegistryRepository.parse_obj(
-                ecr_repo, context=context
-            ).fqn
+                {
+                    "repo_name": v.get("repo_name"),
+                    "registry": ElasticContainerRegistry.parse_obj(
+                        {
+                            "account_id": v.get("account_id"),
+                            "alias": v.get("registry_alias"),
+                            "aws_region": v.get("aws_region"),
+                            "context": values.get("context"),
+                        }
+                    ),
+                }
+            )
+        return v
+
+    @validator("repo", pre=True, always=True, allow_reuse=True)
+    def _set_repo(cls, v: Optional[str], values: Dict[str, Any]) -> Optional[str]:
+        """Set the value of ``repo``."""
+        if v:
+            return v
+
+        ecr_repo: Optional[ElasticContainerRegistryRepository] = values.get("ecr_repo")
+        if ecr_repo:
+            return ecr_repo.fqn
+
         return None
 
+    @validator("dockerfile", pre=True, always=True, allow_reuse=True)
+    def _validate_dockerfile(cls, v: Any, values: Dict[str, Any]) -> Any:
+        """Validate ``dockerfile``."""
+        path: Path = values["path"]
+        dockerfile = path / v
+        if not dockerfile.is_file():
+            raise ValueError(
+                f"Dockerfile does not exist at path provided: {dockerfile}"
+            )
+        return v
+
 
 def build(*, context: CfnginContext, **kwargs: Any) -> DockerHookData:
     """Docker image build hook.
@@ -253,8 +193,7 @@ def build(*, context: CfnginContext, **kwargs: Any) -> DockerHookData:
     kwargs are parsed by :class:`~runway.cfngin.hooks.docker.image.ImageBuildArgs`.
 
     """
-    kwargs.pop("provider", None)  # not needed
-    args = ImageBuildArgs.parse_obj(kwargs, context=context)
+    args = ImageBuildArgs.parse_obj({"context": context, **kwargs})
     docker_hook_data = DockerHookData.from_cfngin_context(context)
     image, logs = cast(
         Tuple[Image, Iterator[Dict[str, str]]],
diff --git a/runway/cfngin/hooks/docker/image/_push.py b/runway/cfngin/hooks/docker/image/_push.py
index 7cc58ec3..7ff86b42 100644
--- a/runway/cfngin/hooks/docker/image/_push.py
+++ b/runway/cfngin/hooks/docker/image/_push.py
@@ -3,12 +3,20 @@
 Replicates the functionality of the ``docker image push`` CLI command.
 
 """
+# pylint: disable=no-self-argument,no-self-use
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
-from ..data_models import BaseModel, DockerImage, ElasticContainerRegistryRepository
+from pydantic import Field, validator
+
+from .....utils import BaseModel
+from ..data_models import (
+    DockerImage,
+    ElasticContainerRegistry,
+    ElasticContainerRegistryRepository,
+)
 from ..hook_data import DockerHookData
 
 if TYPE_CHECKING:
@@ -18,74 +26,76 @@ LOGGER = logging.getLogger(__name__.replace("._", "."))
 
 
 class ImagePushArgs(BaseModel):
-    """Args passed to image.push.
+    """Args passed to image.push."""
+
+    _ctx: Optional[CfnginContext] = Field(default=None, alias="context", export=False)
 
-    Attributes:
-        repo: URI of a non Docker Hub repository where the image will be stored.
-        tags: List of tags to push.
+    ecr_repo: Optional[ElasticContainerRegistryRepository] = None  # depends on _ctx
+    """AWS Elastic Container Registry repository information.
+    Providing this will automatically construct the repo URI.
+    If provided, do not provide ``repo``.
+
+    If using a private registry, only ``repo_name`` is required.
+    If using a public registry, ``repo_name`` and ``registry_alias``.
 
     """
 
-    repo: Optional[str]
-    tags: List[str]
-
-    def __init__(
-        self,
-        *,
-        ecr_repo: Optional[Dict[str, Any]] = None,
-        image: Optional[DockerImage] = None,
-        repo: Optional[str] = None,
-        tags: Optional[List[str]] = None,
-        **kwargs: Any,
-    ) -> None:
-        """Instantiate class.
-
-        Args:
-            ecr_repo: AWS Elastic Container Registry repository information.
-                Providing this will automatically create the repo URI.
-                If provided, do not provide ``repo``.
-            image: Image to push.
-            repo: URI of a non Docker Hub repository where the image will be stored.
-                If providing one of the other repo values, leave this value empty.
-            tags: List of tags to push.
-
-        """
-        super().__init__(**kwargs)
-        self.repo = self.determine_repo(
-            context=self._ctx, ecr_repo=ecr_repo, image=image, repo=repo
-        )
-        if image and not tags:
-            tags = image.tags
-        self.tags = cast(
-            List[str], self._validate_list_str(tags or ["latest"], required=True)
-        )
-
-    @staticmethod
-    def determine_repo(
-        context: Optional[CfnginContext] = None,
-        ecr_repo: Optional[Dict[str, Optional[str]]] = None,
-        image: Optional[DockerImage] = None,
-        repo: Optional[str] = None,
-    ) -> Optional[str]:
-        """Determine repo URI.
-
-        Args:
-            context: CFNgin context.
-            ecr_repo: AWS Elastic Container Registry options.
-            image: Docker image object.
-            repo: URI of a non Docker Hub repository.
-
-        """
-        if repo:
-            return repo
-        if isinstance(image, DockerImage):
+    image: Optional[DockerImage] = None
+    """Image to push."""
+
+    repo: Optional[str] = None  # depends on ecr_repo & image
+    """URI of a non Docker Hub repository where the image will be stored."""
+
+    tags: List[str] = []  # depends on image
+    """List of tags to push."""
+
+    @validator("ecr_repo", pre=True, allow_reuse=True)
+    def _set_ecr_repo(cls, v: Any, values: Dict[str, Any]) -> Any:
+        """Set the value of ``ecr_repo``."""
+        if v and isinstance(v, dict):
+            return ElasticContainerRegistryRepository.parse_obj(
+                {
+                    "repo_name": v.get("repo_name"),
+                    "registry": ElasticContainerRegistry.parse_obj(
+                        {
+                            "account_id": v.get("account_id"),
+                            "alias": v.get("registry_alias"),
+                            "aws_region": v.get("aws_region"),
+                            "context": values.get("context"),
+                        }
+                    ),
+                }
+            )
+        return v
+
+    @validator("repo", pre=True, always=True, allow_reuse=True)
+    def _set_repo(cls, v: Optional[str], values: Dict[str, Any]) -> Optional[str]:
+        """Set the value of ``repo``."""
+        if v:
+            return v
+
+        image: Optional[DockerImage] = values.get("image")
+        if image:
             return image.repo
+
+        ecr_repo: Optional[ElasticContainerRegistryRepository] = values.get("ecr_repo")
         if ecr_repo:
-            return ElasticContainerRegistryRepository.parse_obj(
-                ecr_repo, context=context
-            ).fqn
+            return ecr_repo.fqn
+
         return None
 
+    @validator("tags", pre=True, always=True, allow_reuse=True)
+    def _set_tags(cls, v: List[str], values: Dict[str, Any]) -> List[str]:
+        """Set the value of ``tags``."""
+        if v:
+            return v
+
+        image: Optional[DockerImage] = values.get("image")
+        if image:
+            return image.tags
+
+        return ["latest"]
+
 
 def push(*, context: CfnginContext, **kwargs: Any) -> DockerHookData:
     """Docker image push hook.
@@ -95,8 +105,7 @@ def push(*, context: CfnginContext, **kwargs: Any) -> DockerHookData:
     kwargs are parsed by :class:`~runway.cfngin.hooks.docker.image.ImagePushArgs`.
 
     """
-    kwargs.pop("provider", None)  # not needed
-    args = ImagePushArgs.parse_obj(kwargs, context=context)
+    args = ImagePushArgs.parse_obj({"context": context, **kwargs})
     docker_hook_data = DockerHookData.from_cfngin_context(context)
     LOGGER.info("pushing image %s...", args.repo)
     for tag in args.tags:
diff --git a/runway/cfngin/hooks/docker/image/_remove.py b/runway/cfngin/hooks/docker/image/_remove.py
index 44fc159c..a77abe66 100644
--- a/runway/cfngin/hooks/docker/image/_remove.py
+++ b/runway/cfngin/hooks/docker/image/_remove.py
@@ -3,14 +3,21 @@
 Replicates the functionality of the ``docker image remove`` CLI command.
 
 """
+# pylint: disable=no-self-argument,no-self-use
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, cast
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 from docker.errors import ImageNotFound
-
-from ..data_models import BaseModel, DockerImage, ElasticContainerRegistryRepository
+from pydantic import Field, validator
+
+from .....utils import BaseModel
+from ..data_models import (
+    DockerImage,
+    ElasticContainerRegistry,
+    ElasticContainerRegistryRepository,
+)
 from ..hook_data import DockerHookData
 
 if TYPE_CHECKING:
@@ -20,83 +27,81 @@ LOGGER = logging.getLogger(__name__.replace("._", "."))
 
 
 class ImageRemoveArgs(BaseModel):
-    """Args passed to image.remove.
+    """Args passed to image.remove."""
+
+    _ctx: Optional[CfnginContext] = Field(default=None, alias="context", export=False)
+
+    ecr_repo: Optional[ElasticContainerRegistryRepository] = None  # depends on _ctx
+    """AWS Elastic Container Registry repository information.
+    Providing this will automatically construct the repo URI.
+    If provided, do not provide ``repo``.
 
-    Attributes:
-        force: Whether to force the removal of the image.
-        noprune: Whether to delete untagged parents.
-        repo: URI of a non Docker Hub repository where the image will be stored.
-        tags: List of tags to remove.
+    If using a private registry, only ``repo_name`` is required.
+    If using a public registry, ``repo_name`` and ``registry_alias``.
 
     """
 
-    force: bool
-    noprune: bool
-    repo: Optional[str]
-    tags: List[str]
-
-    def __init__(
-        self,
-        *,
-        ecr_repo: Optional[Dict[str, Any]] = None,
-        force: bool = False,
-        image: DockerImage = None,
-        noprune: bool = False,
-        repo: Optional[str] = None,
-        tags: Optional[List[str]] = None,
-        **kwargs: Any,
-    ) -> None:
-        """Instantiate class.
-
-        Args:
-            ecr_repo: AWS Elastic Container Registry repository information.
-                Providing this will automatically create the repo URI.
-                If provided, do not provide ``repo``.
-            force: Whether to force the removal of the image.
-            image: Image to push.
-            noprune: Whether to delete untagged parents.
-            repo: URI of a non Docker Hub repository where the image will be stored.
-                If providing one of the other repo values, leave this value empty.
-            tags: List of tags to remove.
-
-        """
-        super().__init__(**kwargs)
-        self.force = force
-        self.noprune = noprune
-        self.repo = self.determine_repo(
-            context=self._ctx, ecr_repo=ecr_repo, image=image, repo=repo
-        )
-        if image and not tags:
-            tags = image.tags
-        self.tags = cast(
-            List[str], self._validate_list_str(tags or ["latest"], required=True)
-        )
-
-    @staticmethod
-    def determine_repo(
-        context: Optional[CfnginContext] = None,
-        ecr_repo: Optional[Dict[str, Optional[str]]] = None,
-        image: Optional[DockerImage] = None,
-        repo: Optional[str] = None,
-    ) -> Optional[str]:
-        """Determine repo URI.
-
-        Args:
-            context: CFNgin context.
-            ecr_repo: AWS Elastic Container Registry options.
-            image: Docker image object.
-            repo: URI of a non Docker Hub repository.
-
-        """
-        if repo:
-            return repo
-        if isinstance(image, DockerImage):
+    force: bool = False
+    """Whether to force the removal of the image."""
+
+    image: Optional[DockerImage] = None
+    """Image to push."""
+
+    noprune: bool = False
+    """Whether to delete untagged parents."""
+
+    repo: Optional[str] = None  # depends on ecr_repo & image
+    """URI of a non Docker Hub repository where the image will be stored."""
+
+    tags: List[str] = []  # depends on image
+    """List of tags to remove."""
+
+    @validator("ecr_repo", pre=True, allow_reuse=True)
+    def _set_ecr_repo(cls, v: Any, values: Dict[str, Any]) -> Any:
+        """Set the value of ``ecr_repo``."""
+        if v and isinstance(v, dict):
+            return ElasticContainerRegistryRepository.parse_obj(
+                {
+                    "repo_name": v.get("repo_name"),
+                    "registry": ElasticContainerRegistry.parse_obj(
+                        {
+                            "account_id": v.get("account_id"),
+                            "alias": v.get("registry_alias"),
+                            "aws_region": v.get("aws_region"),
+                            "context": values.get("context"),
+                        }
+                    ),
+                }
+            )
+        return v
+
+    @validator("repo", pre=True, always=True, allow_reuse=True)
+    def _set_repo(cls, v: Optional[str], values: Dict[str, Any]) -> Optional[str]:
+        """Set the value of ``repo``."""
+        if v:
+            return v
+
+        image: Optional[DockerImage] = values.get("image")
+        if image:
             return image.repo
+
+        ecr_repo: Optional[ElasticContainerRegistryRepository] = values.get("ecr_repo")
         if ecr_repo:
-            return ElasticContainerRegistryRepository.parse_obj(
-                ecr_repo, context=context
-            ).fqn
-        raise ValueError("a repo must be specified")
+            return ecr_repo.fqn
+
+        return None
+
+    @validator("tags", pre=True, always=True, allow_reuse=True)
+    def _set_tags(cls, v: List[str], values: Dict[str, Any]) -> List[str]:
+        """Set the value of ``tags``."""
+        if v:
+            return v
+
+        image: Optional[DockerImage] = values.get("image")
+        if image:
+            return image.tags
+
+        return ["latest"]
 
 
 def remove(*, context: CfnginContext, **kwargs: Any) -> DockerHookData:
@@ -107,8 +112,7 @@ def remove(*, context: CfnginContext, **kwargs: Any) -> DockerHookData:
     kwargs are parsed by :class:`~runway.cfngin.hooks.docker.image.ImageRemoveArgs`.
 
     """
-    kwargs.pop("provider", None)  # not needed
-    args = ImageRemoveArgs.parse_obj(kwargs, context=context)
+    args = ImageRemoveArgs.parse_obj({"context": context, **kwargs})
     docker_hook_data = DockerHookData.from_cfngin_context(context)
     LOGGER.info("removing local image %s...", args.repo)
     for tag in args.tags:


Test patch:
diff --git a/tests/unit/cfngin/hooks/docker/image/test_build.py b/tests/unit/cfngin/hooks/docker/image/test_build.py
index d60cdd39..1798b3bd 100644
--- a/tests/unit/cfngin/hooks/docker/image/test_build.py
+++ b/tests/unit/cfngin/hooks/docker/image/test_build.py
@@ -3,16 +3,17 @@
 # pyright: basic
 from __future__ import annotations
 
-from copy import deepcopy
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
 import pytest
 from docker.models.images import Image
 from mock import MagicMock
+from pydantic import ValidationError
 
 from runway.cfngin.hooks.docker.data_models import (
     DockerImage,
+    ElasticContainerRegistry,
     ElasticContainerRegistryRepository,
 )
 from runway.cfngin.hooks.docker.hook_data import DockerHookData
@@ -33,6 +34,14 @@ if TYPE_CHECKING:
 MODULE = "runway.cfngin.hooks.docker.image._build"
 
 
+@pytest.fixture(scope="function")
+def tmp_dockerfile(cd_tmp_path: Path) -> Path:
+    """Create temporary Dockerfile."""
+    dockerfile = cd_tmp_path / "Dockerfile"
+    dockerfile.touch()
+    return dockerfile
+
+
 def test_build(
     cfngin_context: MockCFNginContext, mocker: MockerFixture, tmp_path: Path
 ) -> None:
@@ -71,30 +80,8 @@ def test_build(
 class TestDockerImageBuildApiOptions:
     """Test runway.cfngin.hooks.docker.image._build.DockerImageBuildApiOptions."""
 
-    def test_init(self) -> None:
-        """Test init."""
-        args = {
-            "buildargs": {"key": "val"},
-            "custom_context": True,
-            "extra_hosts": {"host": "local"},
-            "forcerm": True,
-            "isolation": "yes",
-            "network_mode": "host",
-            "nocache": True,
-            "platform": "x86",
-            "pull": True,
-            "rm": False,
-            "squash": True,
-            "tag": "latest",
-            "target": "dev",
-            "timeout": 3,
-            "use_config_proxy": True,
-        }
-        obj = DockerImageBuildApiOptions(**deepcopy(args))
-        assert obj.dict() == args
-
-    def test_init_default(self) -> None:
-        """Test init default."""
+    def test_field_defaults(self) -> None:
+        """Test field defaults."""
         obj = DockerImageBuildApiOptions()
         assert not obj.buildargs and isinstance(obj.buildargs, dict)
         assert obj.custom_context is False
@@ -113,111 +100,90 @@ class TestDockerImageBuildApiOptions:
         assert obj.use_config_proxy is False
 
 
+@pytest.mark.usefixtures("tmp_dockerfile")
 class TestImageBuildArgs:
     """Test runway.cfngin.hooks.docker.image._build.ImageBuildArgs."""
 
-    def test_determine_repo(self) -> None:
-        """Test determine_repo."""
+    @pytest.mark.parametrize(
+        "repo, tag, expected",
+        [(None, None, None), ("foo", None, "foo"), ("foo", "bar", "bar")],
+    )
+    def test__set_docker_dict(
+        self,
+        expected: Optional[str],
+        repo: Optional[str],
+        tag: Optional[str],
+        tmp_path: Path,
+    ) -> None:
+        """Test _set_docker."""
         assert (
-            ImageBuildArgs.determine_repo(
-                context=None, ecr_repo={"key": "val"}, repo="something"
-            )
-            == "something"
+            ImageBuildArgs.parse_obj(
+                {
+                    "docker": {"tag": tag} if tag else {},
+                    "path": tmp_path,
+                    "repo": repo,
+                }
+            ).docker.tag
+            == expected
         )
 
-    def test_determine_repo_ecr(self, mocker: MockerFixture) -> None:
-        """Test determine_repo ecr."""
-        repo = ElasticContainerRegistryRepository(
-            account_id="123456012", aws_region="us-east-1", repo_name="test"
-        )
-        mocker.patch(
-            MODULE + ".ElasticContainerRegistryRepository",
-            parse_obj=MagicMock(return_value=repo),
-        )
+    @pytest.mark.parametrize(
+        "repo, tag, expected",
+        [(None, None, None), ("foo", None, "foo"), ("foo", "bar", "bar")],
+    )
+    def test__set_docker_model(
+        self,
+        expected: Optional[str],
+        repo: Optional[str],
+        tag: Optional[str],
+        tmp_path: Path,
+    ) -> None:
+        """Test _set_docker."""
         assert (
-            ImageBuildArgs.determine_repo(
-                context=None,
-                ecr_repo={
-                    "repo_name": repo.name,
-                    "account_id": repo.registry.account_id,
-                    "aws_region": repo.registry.region,
-                },
-                repo=None,
-            )
-            == repo.fqn
+            ImageBuildArgs(
+                docker=DockerImageBuildApiOptions(tag=tag),
+                path=tmp_path,
+                repo=repo,
+            ).docker.tag
+            == expected
         )
 
-    def test_init(self, mocker: MockerFixture, tmp_path: Path) -> None:
-        """Test init."""
+    def test__set_ecr_repo_from_dict(self, tmp_path: Path) -> None:
+        """Test _set_ecr_repo from Dict."""
         args = {
-            "docker": {"pull": True},
-            "dockerfile": "./dir/Dockerfile",
-            "ecr_repo": {"name": "test"},
-            "path": tmp_path,
-            "repo": "ecr",
-            "tags": ["oldest"],
+            "repo_name": "foo",
+            "account_id": "123",
+            "registry_alias": "bar",
+            "aws_region": "us-west-2",
         }
-        context = MagicMock()
-        mock_validate_dockerfile = mocker.patch.object(
-            ImageBuildArgs, "_validate_dockerfile", return_value=args["dockerfile"]
-        )
-        mock_determine_repo = mocker.patch.object(
-            ImageBuildArgs, "determine_repo", return_value="repo"
-        )
-        obj = ImageBuildArgs.parse_obj(args, context=context)
-        assert obj.path == args["path"]
-        mock_validate_dockerfile.assert_called_once_with(tmp_path, args["dockerfile"])
-        assert obj.dockerfile == args["dockerfile"]
-        mock_determine_repo.assert_called_once_with(
-            context=context, ecr_repo=args["ecr_repo"], repo=args["repo"]
-        )
-        assert obj.repo == mock_determine_repo.return_value
-        assert obj.tags == args["tags"]
-        assert isinstance(obj.docker, DockerImageBuildApiOptions)
-        assert obj.docker.tag == mock_determine_repo.return_value
-
-    def test_init_default(self, mocker: MockerFixture) -> None:
-        """Test init default values."""
-        context = MagicMock()
-        mock_validate_dockerfile = mocker.patch.object(
-            ImageBuildArgs, "_validate_dockerfile", return_value="./Dockerfile"
-        )
-        obj = ImageBuildArgs(context=context)
-        assert obj.path == Path.cwd()
-        mock_validate_dockerfile.assert_called_once_with(Path.cwd(), "./Dockerfile")
-        assert obj.dockerfile == mock_validate_dockerfile.return_value
-        assert not obj.repo
-        assert obj.tags == ["latest"]
-        assert isinstance(obj.docker, DockerImageBuildApiOptions)
-
-    def test_validate_dockerfile(self, tmp_path: Path) -> None:
-        """Test _validate_dockerfile."""
-        (tmp_path / "Dockerfile").touch()
-        assert (
-            ImageBuildArgs._validate_dockerfile(tmp_path, "./Dockerfile")
-            == "./Dockerfile"
-        )
-
-    def test_validate_dockerfile_does_not_exist(self, tmp_path: Path) -> None:
-        """Test _validate_dockerfile does not exist."""
-        with pytest.raises(ValueError) as excinfo:
-            ImageBuildArgs._validate_dockerfile(tmp_path, "./Dockerfile")
-        assert str(excinfo.value).startswith("Dockerfile does not exist at path")
-
-    def test_validate_dockerfile_path_is_dockerfile(self, tmp_path: Path) -> None:
-        """Test _validate_dockerfile does not exist."""
-        path = tmp_path / "Dockerfile"
-        path.touch()
-        with pytest.raises(ValueError) as excinfo:
-            ImageBuildArgs._validate_dockerfile(path, "./Dockerfile")
-        assert str(excinfo.value).startswith(
-            "ImageBuildArgs.path should not reference the Dockerfile directly"
+        obj = ImageBuildArgs.parse_obj({"path": tmp_path, "ecr_repo": args})
+        assert obj.ecr_repo
+        assert obj.ecr_repo.name == args["repo_name"]
+        assert obj.ecr_repo.registry.account_id == args["account_id"]
+        assert obj.ecr_repo.registry.alias == args["registry_alias"]
+        assert obj.ecr_repo.registry.region == args["aws_region"]
+
+    def test__set_repo(self, tmp_path: Path) -> None:
+        """Test _set_repo."""
+        assert ImageBuildArgs(path=tmp_path, repo="something").repo == "something"
+
+    def test__set_repo_ecr(self, tmp_path: Path) -> None:
+        """Test _set_repo ECR."""
+        repo = ElasticContainerRegistryRepository(
+            repo_name="test",
+            registry=ElasticContainerRegistry(
+                account_id="123456789012", aws_region="us-east-1"
+            ),
         )
-
-    def test_validate_dockerfile_path_is_zipfile(self, tmp_path: Path) -> None:
-        """Test _validate_dockerfile path is zipfile."""
-        path = tmp_path / "something.zip"
-        path.touch()
-        assert (
-            ImageBuildArgs._validate_dockerfile(path, "./Dockerfile") == "./Dockerfile"
+        assert ImageBuildArgs(path=tmp_path, ecr_repo=repo).repo == repo.fqn
+
+    def test__validate_dockerfile_raise_value_error(self, tmp_path: Path) -> None:
+        """Test _validate_dockerfile raise ValueError."""
+        with pytest.raises(ValidationError) as excinfo:
+            assert ImageBuildArgs(dockerfile="invalid", path=tmp_path, repo="something")
+        errors = excinfo.value.errors()
+        assert len(errors) == 1
+        assert errors[0]["loc"] == ("dockerfile",)
+        assert errors[0]["msg"].startswith(
+            "Dockerfile does not exist at path provided: "
         )
diff --git a/tests/unit/cfngin/hooks/docker/image/test_push.py b/tests/unit/cfngin/hooks/docker/image/test_push.py
index 5d702064..26144b87 100644
--- a/tests/unit/cfngin/hooks/docker/image/test_push.py
+++ b/tests/unit/cfngin/hooks/docker/image/test_push.py
@@ -5,10 +5,12 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-from mock import MagicMock, call
+from docker.models.images import Image
+from mock import call
 
 from runway.cfngin.hooks.docker.data_models import (
     DockerImage,
+    ElasticContainerRegistry,
     ElasticContainerRegistryRepository,
 )
 from runway.cfngin.hooks.docker.hook_data import DockerHookData
@@ -52,81 +54,49 @@ def test_push(
 class TestImagePushArgs:
     """Test runway.cfngin.hooks.docker.image._push.ImagePushArgs."""
 
-    def test_determine_repo(self) -> None:
-        """Test determine_repo."""
-        assert (
-            ImagePushArgs.determine_repo(
-                context=None, ecr_repo={"key": "val"}, image=None, repo="something"
-            )
-            == "something"
-        )
-
-    def test_determine_repo_ecr(self, mocker: MockerFixture) -> None:
-        """Test determine_repo ecr."""
+    def test__set_ecr_repo_from_dict(self) -> None:
+        """Test _set_ecr_repo from Dict."""
+        args = {
+            "repo_name": "foo",
+            "account_id": "123",
+            "registry_alias": "bar",
+            "aws_region": "us-west-2",
+        }
+        obj = ImagePushArgs.parse_obj({"ecr_repo": args})
+        assert obj.ecr_repo
+        assert obj.ecr_repo.name == args["repo_name"]
+        assert obj.ecr_repo.registry.account_id == args["account_id"]
+        assert obj.ecr_repo.registry.alias == args["registry_alias"]
+        assert obj.ecr_repo.registry.region == args["aws_region"]
+
+    def test__set_repo(self) -> None:
+        """Test _set_repo."""
+        assert ImagePushArgs(repo="something").repo == "something"
+        assert not ImagePushArgs().repo
+
+    def test__set_repo_ecr(self) -> None:
+        """Test _set_repo ECR."""
         repo = ElasticContainerRegistryRepository(
-            account_id="123456012", aws_region="us-east-1", repo_name="test"
-        )
-        mocker.patch(
-            MODULE + ".ElasticContainerRegistryRepository",
-            parse_obj=MagicMock(return_value=repo),
-        )
-        assert (
-            ImagePushArgs.determine_repo(
-                context=None,
-                ecr_repo={
-                    "repo_name": repo.name,
-                    "account_id": repo.registry.account_id,
-                    "aws_region": repo.registry.region,
-                },
-                image=None,
-                repo=None,
-            )
-            == repo.fqn
-        )
-
-    def test_determine_repo_image(self) -> None:
-        """Test determine_repo Image."""
-        repo = "dkr.test.com/image"
-        mock_image = MagicMock(spec=DockerImage, repo=repo)
-        assert (
-            ImagePushArgs.determine_repo(
-                context=None, ecr_repo={"key": "val"}, image=mock_image, repo=None
-            )
-            == repo
-        )
-
-    def test_determine_repo_none(self) -> None:
-        """Test determine_repo None."""
-        assert not ImagePushArgs.determine_repo(
-            context=None, ecr_repo={}, image=None, repo=None
-        )
-
-    def test_init_default(self, mocker: MockerFixture) -> None:
-        """Test init default values."""
-        mock_determine_repo = mocker.patch.object(
-            ImagePushArgs, "determine_repo", return_value="dkr.test.com/image"
-        )
-        obj = ImagePushArgs()
-        mock_determine_repo.assert_called_once_with(
-            context=None, ecr_repo=None, image=None, repo=None
+            repo_name="test",
+            registry=ElasticContainerRegistry(
+                account_id="123456789012", aws_region="us-east-1"
+            ),
         )
-        assert obj.repo == mock_determine_repo.return_value
-        assert obj.tags == ["latest"]
-
-    def test_init_image(self, mocker: MockerFixture) -> None:
-        """Test init Image."""
-        repo = "dkr.test.com/image"
-        tags = ["latest", "oldest"]
-        mock_image = MagicMock(spec=DockerImage, repo=repo, tags=tags)
-        mock_determine_repo = mocker.patch.object(
-            ImagePushArgs, "determine_repo", return_value=repo
-        )
-        obj = ImagePushArgs(image=mock_image)
-        mock_determine_repo.assert_called_once_with(
-            context=None, ecr_repo=None, image=mock_image, repo=None
-        )
-        assert obj.repo == repo
-        assert obj.tags == tags
-
-        # ensure tags are not overwritten if provided
-        assert ImagePushArgs(image=mock_image, tags=["oldest"]).tags == ["oldest"]
+        assert ImagePushArgs(ecr_repo=repo).repo == repo.fqn
+
+    def test__set_repo_image(self, mocker: MockerFixture) -> None:
+        """Test _set_repo image."""
+        mocker.patch.object(Image, "reload")
+        image = DockerImage(image=Image({"RepoTags": ["foo:latest"]}))
+        assert ImagePushArgs(image=image).repo == image.repo
+
+    def test__set_tags(self) -> None:
+        """Test _set_tags."""
+        assert ImagePushArgs(tags=["foo"]).tags == ["foo"]
+        assert ImagePushArgs().tags == ["latest"]
+
+    def test__set_tags_image(self, mocker: MockerFixture) -> None:
+        """Test _set_tags image."""
+        mocker.patch.object(Image, "reload")
+        image = DockerImage(image=Image({"RepoTags": ["foo:bar"]}))
+        assert ImagePushArgs(image=image).tags == image.tags
diff --git a/tests/unit/cfngin/hooks/docker/image/test_remove.py b/tests/unit/cfngin/hooks/docker/image/test_remove.py
index 277584bd..ae24d38d 100644
--- a/tests/unit/cfngin/hooks/docker/image/test_remove.py
+++ b/tests/unit/cfngin/hooks/docker/image/test_remove.py
@@ -5,12 +5,13 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-import pytest
 from docker.errors import ImageNotFound
-from mock import MagicMock, call
+from docker.models.images import Image
+from mock import call
 
 from runway.cfngin.hooks.docker.data_models import (
     DockerImage,
+    ElasticContainerRegistry,
     ElasticContainerRegistryRepository,
 )
 from runway.cfngin.hooks.docker.hook_data import DockerHookData
@@ -34,13 +35,12 @@ def test_remove(
     """Test remove."""
     repo = "dkr.test.com/image"
     tags = ["latest", "oldest"]
-    mock_image = MagicMock(spec=DockerImage, tags=[f"{repo}:{tag}" for tag in tags])
-    mock_image.attrs = {"RepoTags": mock_image.tags}
-    args = ImageRemoveArgs(force=True, image=mock_image, tags=["latest", "oldest"])
+    image = DockerImage(image=Image({"RepoTags": [f"{repo}:{tag}" for tag in tags]}))
+    args = ImageRemoveArgs(force=True, image=image, tags=["latest", "oldest"])
     mocker.patch.object(ImageRemoveArgs, "parse_obj", return_value=args)
     mocker.patch.object(DockerHookData, "client", mock_docker_client)
     docker_hook_data = DockerHookData()
-    docker_hook_data.image = mock_image
+    docker_hook_data.image = image
     mock_from_cfngin_context = mocker.patch.object(
         DockerHookData, "from_cfngin_context", return_value=docker_hook_data
     )
@@ -49,9 +49,7 @@ def test_remove(
     )
     cfngin_context.hook_data["docker"] = docker_hook_data
     assert (
-        remove(
-            context=cfngin_context, force=args.force, image=mock_image, tags=args.tags
-        )
+        remove(context=cfngin_context, force=args.force, image=image, tags=args.tags)
         == docker_hook_data
     )
     mock_from_cfngin_context.assert_called_once_with(cfngin_context)
@@ -99,86 +97,49 @@ def test_remove_image_not_found(
 class TestImageRemoveArgs:
     """Test runway.cfngin.hooks.docker.image._remove.ImageRemoveArgs."""
 
-    def test_determine_repo(self) -> None:
-        """Test determine_repo."""
-        assert (
-            ImageRemoveArgs.determine_repo(
-                context=None, ecr_repo={"key": "val"}, image=None, repo="something"
-            )
-            == "something"
-        )
-
-    def test_determine_repo_ecr(self, mocker: MockerFixture) -> None:
-        """Test determine_repo ecr."""
+    def test__set_ecr_repo_from_dict(self) -> None:
+        """Test _set_ecr_repo from Dict."""
+        args = {
+            "repo_name": "foo",
+            "account_id": "123",
+            "registry_alias": "bar",
+            "aws_region": "us-west-2",
+        }
+        obj = ImageRemoveArgs.parse_obj({"ecr_repo": args})
+        assert obj.ecr_repo
+        assert obj.ecr_repo.name == args["repo_name"]
+        assert obj.ecr_repo.registry.account_id == args["account_id"]
+        assert obj.ecr_repo.registry.alias == args["registry_alias"]
+        assert obj.ecr_repo.registry.region == args["aws_region"]
+
+    def test__set_repo(self) -> None:
+        """Test _set_repo."""
+        assert ImageRemoveArgs(repo="something").repo == "something"
+        assert not ImageRemoveArgs().repo
+
+    def test__set_repo_ecr(self) -> None:
+        """Test _set_repo ECR."""
         repo = ElasticContainerRegistryRepository(
-            account_id="123456012", aws_region="us-east-1", repo_name="test"
-        )
-        mocker.patch(
-            MODULE + ".ElasticContainerRegistryRepository",
-            parse_obj=MagicMock(return_value=repo),
-        )
-        assert (
-            ImageRemoveArgs.determine_repo(
-                context=None,
-                ecr_repo={
-                    "repo_name": repo.name,
-                    "account_id": repo.registry.account_id,
-                    "aws_region": repo.registry.region,
-                },
-                image=None,
-                repo=None,
-            )
-            == repo.fqn
-        )
-
-    def test_determine_repo_image(self) -> None:
-        """Test determine_repo Image."""
-        repo = "dkr.test.com/image"
-        mock_image = MagicMock(spec=DockerImage, repo=repo)
-        assert (
-            ImageRemoveArgs.determine_repo(
-                context=None, ecr_repo={"key": "val"}, image=mock_image, repo=None
-            )
-            == repo
+            repo_name="test",
+            registry=ElasticContainerRegistry(
+                account_id="123456789012", aws_region="us-east-1"
+            ),
         )
-
-    def test_determine_repo_none(self) -> None:
-        """Test determine_repo None."""
-        with pytest.raises(ValueError):
-            ImageRemoveArgs.determine_repo(
-                context=None, ecr_repo={}, image=None, repo=None
-            )
-
-    def test_init_default(self, mocker: MockerFixture) -> None:
-        """Test init default values."""
-        mock_determine_repo = mocker.patch.object(
-            ImageRemoveArgs, "determine_repo", return_value="dkr.test.com/image"
-        )
-        obj = ImageRemoveArgs()
-        mock_determine_repo.assert_called_once_with(
-            context=None, ecr_repo=None, image=None, repo=None
-        )
-        assert obj.force is False
-        assert obj.noprune is False
-        assert obj.repo == mock_determine_repo.return_value
-        assert obj.tags == ["latest"]
-
-    def test_init_image(self, mocker: MockerFixture) -> None:
-        """Test init Image."""
-        repo = "dkr.test.com/image"
-        tags = ["latest", "oldest"]
-        mock_image = MagicMock(spec=DockerImage, repo=repo, tags=tags)
-        mock_determine_repo = mocker.patch.object(
-            ImageRemoveArgs, "determine_repo", return_value=repo
-        )
-        obj = ImageRemoveArgs(force=True, image=mock_image, noprune=True)
-        mock_determine_repo.assert_called_once_with(
-            context=None, ecr_repo=None, image=mock_image, repo=None
-        )
-        assert obj.force is True
-        assert obj.noprune is True
-        assert obj.repo == repo
-        assert obj.tags == tags
-
-        # ensure tags are not overwritten if provided
-        assert ImageRemoveArgs(image=mock_image, tags=["oldest"]).tags == ["oldest"]
+        assert ImageRemoveArgs(ecr_repo=repo).repo == repo.fqn
+
+    def test__set_repo_image(self, mocker: MockerFixture) -> None:
+        """Test _set_repo image."""
+        mocker.patch.object(Image, "reload")
+        image = DockerImage(image=Image({"RepoTags": ["foo:latest"]}))
+        assert ImageRemoveArgs(image=image).repo == image.repo
+
+    def test__set_tags(self) -> None:
+        """Test _set_tags."""
+        assert ImageRemoveArgs(tags=["foo"]).tags == ["foo"]
+        assert ImageRemoveArgs().tags == ["latest"]
+
+    def test__set_tags_image(self, mocker: MockerFixture) -> None:
+        """Test _set_tags image."""
+        mocker.patch.object(Image, "reload")
+        image = DockerImage(image=Image({"RepoTags": ["foo:bar"]}))
+        assert ImageRemoveArgs(image=image).tags == image.tags
diff --git a/tests/unit/cfngin/hooks/docker/test_data_models.py b/tests/unit/cfngin/hooks/docker/test_data_models.py
index af5e9811..4213b346 100644
--- a/tests/unit/cfngin/hooks/docker/test_data_models.py
+++ b/tests/unit/cfngin/hooks/docker/test_data_models.py
@@ -3,14 +3,14 @@
 # pyright: basic
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import TYPE_CHECKING
 
 import pytest
 from docker.models.images import Image
 from mock import MagicMock
+from pydantic import ValidationError
 
 from runway.cfngin.hooks.docker.data_models import (
-    BaseModel,
     DockerImage,
     ElasticContainerRegistry,
     ElasticContainerRegistryRepository,
@@ -18,10 +18,6 @@ from runway.cfngin.hooks.docker.data_models import (
 from runway.utils import MutableMap
 
 if TYPE_CHECKING:
-    from pathlib import Path
-
-    from pytest_mock import MockerFixture
-
     from ....factories import MockCFNginContext
 
 MODULE = "runway.cfngin.hooks.docker.data_models"
@@ -40,249 +36,6 @@ def mock_image() -> MagicMock:
     return MagicMock(spec=Image, **MOCK_IMAGE_PROPS)
 
 
-class SampleModel(BaseModel):
-    """Class to test the BaseModel."""
-
-    def __init__(  # pylint: disable=super-init-not-called
-        self, *_: Any, **kwargs: Any
-    ) -> None:
-        """Instantiate class."""
-        for k, v in kwargs.items():
-            setattr(self, k, v)
-
-
-class TestBaseModel:
-    """Test runway.cfngin.hooks.docker._data_models.BaseModel."""
-
-    def test_dict(self) -> None:
-        """Test dict."""
-        obj = SampleModel(key="val", _key="val")
-        assert obj.dict() == {"key": "val"}
-
-    def test_find(self) -> None:
-        """Test find."""
-        obj = SampleModel(key="val")
-        assert obj.find("key") == "val"
-        assert obj.find("missing") is None
-        assert obj.find("missing", "default") == "default"
-
-    def test_find_nested(self) -> None:
-        """Test find nested value."""
-        assert (
-            SampleModel(lang=SampleModel(python="found")).find("lang.python") == "found"
-        )
-        assert (
-            SampleModel(lang=SampleModel(**{"python3.8": "found"})).find(
-                "lang.python3.8"
-            )
-            == "found"
-        )
-        assert (
-            SampleModel(lang=SampleModel(**{"python3.8": "found"})).find(
-                "lang.python3.9", "not found"
-            )
-            == "not found"
-        )
-
-    def test_find_nested_attr_key_error(self) -> None:
-        """Test find nested AttributeError/KeyError."""
-        assert (
-            SampleModel(lang={"python": {"3": "found"}}).find(
-                "lang.python.3", "default"
-            )
-            == "default"
-        )
-
-    def test_get(self) -> None:
-        """Test get."""
-        obj = SampleModel(key="val")
-        assert obj.get("key") == obj["key"]
-        assert obj.get("missing") is None
-        assert obj.get("missing", "default") == "default"
-
-    @pytest.mark.parametrize("provided", [{"key": "val"}, SampleModel(key="val")])
-    def test_parse_obj(self, provided: Any) -> None:
-        """Test parse_obj."""
-        obj = SampleModel.parse_obj(provided)
-        provided["context"] = None
-        assert isinstance(obj, SampleModel)
-        assert obj == provided
-
-    def test_parse_obj_type_error(self) -> None:
-        """Test parse_obj raise TypeError."""
-        with pytest.raises(TypeError):
-            SampleModel.parse_obj(["something"])
-
-    @pytest.mark.parametrize(
-        "provided, expected",
-        [(True, True), (False, False), (None, False), ("", False), ("false", True)],
-    )
-    def test_validate_bool(self, provided: Any, expected: bool) -> None:
-        """Test _validate_bool."""
-        assert BaseModel._validate_bool(provided) is expected
-
-    @pytest.mark.parametrize(
-        "provided, optional, required, expected",
-        [
-            (None, False, False, {}),
-            (None, True, False, None),
-            ({}, False, False, {}),
-            ({}, True, False, {}),
-            ({"key": "val"}, False, False, {"key": "val"}),
-            ({"key": "val"}, True, False, {"key": "val"}),
-            ({"key": "val"}, False, True, {"key": "val"}),
-            ({"key": "val"}, True, True, {"key": "val"}),
-            (SampleModel(key="val"), False, False, {"key": "val"}),
-        ],
-    )
-    def test_validate_dict(
-        self,
-        provided: Any,
-        optional: bool,
-        required: bool,
-        expected: Optional[Dict[str, Any]],
-    ) -> None:
-        """Test _validate_dict."""
-        assert (
-            BaseModel._validate_dict(provided, optional=optional, required=required)
-            == expected
-        )
-
-    def test_validate_dict_value_error(self) -> None:
-        """Test _validate_dict raise ValueError."""
-        with pytest.raises(ValueError):
-            BaseModel._validate_dict(["something"])  # type: ignore
-        with pytest.raises(ValueError):
-            BaseModel._validate_dict(None, required=True)
-        with pytest.raises(ValueError):
-            BaseModel._validate_dict({}, required=True)
-
-    @pytest.mark.parametrize(
-        "provided, optional, required, expected",
-        [
-            (None, False, False, 0),
-            (None, True, False, None),
-            (0, False, False, 0),
-            (0, True, False, 0),
-            (0, False, True, 0),
-            (0, True, True, 0),
-            (13, False, False, 13),
-            ("13", False, False, 13),
-        ],
-    )
-    def test_validate_int(
-        self, provided: Any, optional: bool, required: bool, expected: Optional[int]
-    ) -> None:
-        """Test _validate_int."""
-        assert (
-            BaseModel._validate_int(provided, optional=optional, required=required)
-            == expected
-        )
-
-    def test_validate_int_value_error(self) -> None:
-        """Test _validate_int raise ValueError."""
-        with pytest.raises(ValueError):
-            BaseModel._validate_int(None, required=True)
-        with pytest.raises(ValueError):
-            BaseModel._validate_int("something")
-
-    @pytest.mark.parametrize(
-        "provided, optional, required, expected",
-        [
-            (None, False, False, []),
-            (None, True, False, None),
-            ([], False, False, []),
-            ([], True, False, []),
-            (["something"], False, True, ["something"]),
-            (["something"], True, True, ["something"]),
-            ({"something"}, False, False, ["something"]),
-            (("something",), False, False, ["something"]),
-            ("abc", False, False, ["a", "b", "c"]),
-        ],
-    )
-    def test_validate_list_str(
-        self,
-        provided: Any,
-        optional: bool,
-        required: bool,
-        expected: Optional[List[str]],
-    ) -> None:
-        """Test _validate_list_str."""
-        assert (
-            BaseModel._validate_list_str(provided, optional=optional, required=required)
-            == expected
-        )
-
-    def test_validate_list_str_type_error(self) -> None:
-        """Test _validate_list_str raise TypeError."""
-        with pytest.raises(TypeError):
-            BaseModel._validate_list_str(["something", None])
-        with pytest.raises(TypeError):
-            BaseModel._validate_list_str(1)
-
-    def test_validate_list_str_value_error(self) -> None:
-        """Test _validate_list_str raise ValueError."""
-        with pytest.raises(ValueError):
-            BaseModel._validate_list_str(None, required=True)
-
-    def test_validate_path(self, tmp_path: Path) -> None:
-        """Test validate_path."""
-        assert BaseModel._validate_path(str(tmp_path)) == tmp_path
-        assert BaseModel._validate_path(str(tmp_path), must_exist=True) == tmp_path
-        assert BaseModel._validate_path(tmp_path) == tmp_path
-        assert BaseModel._validate_path(tmp_path, must_exist=True) == tmp_path
-
-    def test_validate_path_type_error(self) -> None:
-        """Test _validate_path raise TypeError."""
-        with pytest.raises(TypeError):
-            BaseModel._validate_path(13)
-
-    def test_validate_path_value_error(self, tmp_path: Path) -> None:
-        """Test _validate_path raise ValueError."""
-        with pytest.raises(ValueError):
-            BaseModel._validate_path(tmp_path / "missing", must_exist=True)
-
-    @pytest.mark.parametrize(
-        "provided, optional, required, expected",
-        [
-            (None, False, False, "None"),
-            (None, True, False, None),
-            ("", False, False, ""),
-            ("", True, False, ""),
-            ("something", False, False, "something"),
-            ("something", True, False, "something"),
-            ("something", False, True, "something"),
-            ("something", True, True, "something"),
-            (0, False, False, "0"),
-            (0, True, False, None),
-            (1, False, False, "1"),
-        ],
-    )
-    def test_validate_str(
-        self, provided: Any, optional: bool, required: bool, expected: Optional[str]
-    ) -> None:
-        """Test _validate_str."""
-        assert (
-            BaseModel._validate_str(provided, optional=optional, required=required)
-            == expected
-        )
-
-    @pytest.mark.parametrize(
-        "provided", [{"key": "val"}, ["something"], {"something"}, ("something",)]
-    )
-    def test_validate_str_type_error(self, provided: Any) -> None:
-        """Test _validate_str raise TypeError."""
-        with pytest.raises(TypeError):
-            BaseModel._validate_str(provided)
-
-    def test_validate_str_value_error(self) -> None:
-        """Test _validate_str raise ValueError."""
-        with pytest.raises(ValueError):
-            BaseModel._validate_str(None, required=True)
-        with pytest.raises(ValueError):
-            BaseModel._validate_str("", required=True)
-
-
 class TestDockerImage:
     """Test runway.cfngin.hooks.docker.data_models.DockerImage."""
 
@@ -290,22 +43,16 @@ class TestDockerImage:
         """Test id."""
         obj = DockerImage(image=mock_image)
         assert obj.id == MOCK_IMAGE_PROPS["id"]
-        obj.id = "new-id"
-        assert obj.id == "new-id"
 
     def test_repo(self, mock_image: MagicMock) -> None:
         """Test repo."""
         obj = DockerImage(image=mock_image)
         assert obj.repo == MOCK_IMAGE_REPO
-        obj.repo = "new-repo"
-        assert obj.repo == "new-repo"
 
     def test_sort_id(self, mock_image: MagicMock) -> None:
         """Test short_id."""
         obj = DockerImage(image=mock_image)
         assert obj.short_id == MOCK_IMAGE_PROPS["short_id"]
-        obj.short_id = "new-id"
-        assert obj.short_id == "new-id"
 
     def test_tags(self, mock_image: MagicMock) -> None:
         """Test tags."""
@@ -356,9 +103,12 @@ class TestElasticContainerRegistry:
 
     def test_init_no_context(self) -> None:
         """Test init with no context."""
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ValidationError) as excinfo:
             ElasticContainerRegistry()
-        assert str(excinfo.value) == "context is required to resolve values"
+        errors = excinfo.value.errors()
+        assert len(errors) == 1
+        assert errors[0]["loc"] == ("__root__",)
+        assert errors[0]["msg"] == "context is required to resolve values"
 
     def test_init_private(self) -> None:
         """Test init private."""
@@ -382,49 +132,15 @@ class TestElasticContainerRegistry:
 class TestElasticContainerRegistryRepository:
     """Test runway.cfngin.hooks.docker._data_models.ElasticContainerRegistryRepository."""
 
-    def test_fqn_private(
-        self, cfngin_context: MockCFNginContext, mocker: MockerFixture
-    ) -> None:
+    def test_fqn(self, cfngin_context: MockCFNginContext) -> None:
         """Test init private."""
         account_id = "123456789012"
         region = "us-east-1"
-        mock_registry = mocker.patch(
-            MODULE + ".ElasticContainerRegistry", MagicMock(fqn="repository/")
-        )
-        mock_registry.return_value = mock_registry
-        obj = ElasticContainerRegistryRepository(
-            repo_name="something",
-            account_id=account_id,
-            aws_region=region,
-            context=cfngin_context,
-        )
-        assert obj.fqn == "repository/something"
-        mock_registry.assert_called_once_with(
-            account_id=account_id, alias=None, aws_region=region, context=cfngin_context
-        )
 
-    def test_fqn_public(
-        self, cfngin_context: MockCFNginContext, mocker: MockerFixture
-    ) -> None:
-        """Test init public."""
-        mock_registry = mocker.patch(
-            MODULE + ".ElasticContainerRegistry", MagicMock(fqn="repository/")
-        )
-        mock_registry.return_value = mock_registry
         obj = ElasticContainerRegistryRepository(
-            repo_name="something", registry_alias="test", context=cfngin_context
-        )
-        assert obj.fqn == "repository/something"
-        mock_registry.assert_called_once_with(
-            account_id=None, alias="test", aws_region=None, context=cfngin_context
-        )
-
-    def test_init_default(self, mocker: MockerFixture) -> None:
-        """Test init default values."""
-        mock_registry = mocker.patch(MODULE + ".ElasticContainerRegistry")
-        obj = ElasticContainerRegistryRepository(repo_name="something")
-        assert obj.name == "something"
-        assert obj.registry == mock_registry.return_value
-        mock_registry.assert_called_once_with(
-            account_id=None, alias=None, aws_region=None, context=None
+            repo_name="something",
+            registry=ElasticContainerRegistry(
+                account_id=account_id, aws_region=region, context=cfngin_context
+            ),
         )
+        assert obj.fqn == f"{obj.registry.fqn}{obj.name}"
diff --git a/tests/unit/cfngin/hooks/docker/test_login.py b/tests/unit/cfngin/hooks/docker/test_login.py
index f8be4495..52413dc0 100644
--- a/tests/unit/cfngin/hooks/docker/test_login.py
+++ b/tests/unit/cfngin/hooks/docker/test_login.py
@@ -4,9 +4,9 @@
 from __future__ import annotations
 
 from copy import deepcopy
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 
-from mock import MagicMock
+import pytest
 
 from runway.cfngin.hooks.docker import login
 from runway.cfngin.hooks.docker._login import LoginArgs
@@ -49,72 +49,54 @@ def test_login(
 class TestLoginArgs:
     """Test runway.cfngin.hooks.docker._login.LoginArgs."""
 
-    def test_determine_registry(self) -> None:
-        """Test determine_registry."""
+    def test__set_ecr(self, mocker: MockerFixture) -> None:
+        """Test _set_ecr."""
+        expected = ElasticContainerRegistry(alias="foobar")
+        mock_parse_obj = mocker.patch.object(
+            ElasticContainerRegistry, "parse_obj", return_value=expected
+        )
         assert (
-            LoginArgs.determine_registry(context=None, ecr=True, registry="something")
-            == "something"
+            LoginArgs.parse_obj(
+                {"ecr": expected.dict(), "password": "", "username": ""}
+            ).ecr
+            == expected
         )
+        mock_parse_obj.assert_called_once_with({"context": None, **expected.dict()})
 
-    def test_determine_registry_ecr(self, mocker: MockerFixture) -> None:
-        """Test determine_registry ecr."""
-        registry = ElasticContainerRegistry(
-            account_id="123456012", aws_region="us-east-1"
-        )
-        mocker.patch(
-            MODULE + ".ElasticContainerRegistry",
-            parse_obj=MagicMock(return_value=registry),
-        )
+    @pytest.mark.parametrize(
+        "ecr, registry, expected",
+        [
+            (None, "something", "something"),
+            (None, None, None),
+            (ElasticContainerRegistry(alias="foobar"), "something", "something"),
+            (
+                ElasticContainerRegistry(alias="foobar"),
+                None,
+                ElasticContainerRegistry.PUBLIC_URI_TEMPLATE.format(
+                    registry_alias="foobar"
+                ),
+            ),
+        ],
+    )
+    def test__set_registry(
+        self,
+        ecr: Optional[ElasticContainerRegistry],
+        expected: Optional[str],
+        registry: Optional[str],
+    ) -> None:
+        """Test _set_registry."""
         assert (
-            LoginArgs.determine_registry(context=None, ecr=True, registry=None)
-            == registry.fqn
+            LoginArgs(ecr=ecr, password="", registry=registry, username="").registry
+            == expected
         )
 
-    def test_init_default(self) -> None:
-        """Test init defalt."""
+    def test_field_defaults(self) -> None:
+        """Test field defalts."""
         args = {"password": "p@ssword", "username": "test-user"}
         obj = LoginArgs.parse_obj(deepcopy(args))
         assert not obj.dockercfg_path
+        assert not obj.ecr
         assert not obj.email
         assert obj.password == args["password"]
         assert not obj.registry
         assert obj.username == args["username"]
-
-    def test_init_ecr(self, mocker: MockerFixture) -> None:
-        """Test init ecr."""
-        context = MagicMock()
-        registry = ElasticContainerRegistry(
-            account_id="123456012", aws_region="us-east-1"
-        )
-        mock_determine_registry = mocker.patch.object(
-            LoginArgs, "determine_registry", return_value=registry.fqn
-        )
-        args = {
-            "ecr": {"account_id": "123456012", "aws_region": "us-east-1"},
-            "password": "p@ssword",
-        }
-        obj = LoginArgs.parse_obj(deepcopy(args), context=context)
-        mock_determine_registry.assert_called_once_with(
-            context=context, ecr=args["ecr"], registry=None
-        )
-        assert not obj.dockercfg_path
-        assert not obj.email
-        assert obj.password == args["password"]
-        assert obj.registry == registry.fqn
-        assert obj.username == "AWS"
-
-    def test_init_other_registry(self) -> None:
-        """Test init with "other" registry."""
-        args = {
-            "dockercfg_path": "./.docker/config.json",
-            "email": "user@test.com",
-            "password": "p@ssword",
-            "username": "test-user",
-            "registry": "dkr.test.com",
-        }
-        obj = LoginArgs.parse_obj(deepcopy(args))
-        assert obj.dockercfg_path == args["dockercfg_path"]
-        assert obj.email == args["email"]
-        assert obj.password == args["password"]
-        assert obj.registry == args["registry"]
-        assert obj.username == args["username"]