diff --git a/.flake8 b/.flake8
index e16565edc6..cef44ef306 100644
--- a/.flake8
+++ b/.flake8
@@ -45,6 +45,7 @@ fcn_exclude_functions =
     yaml,
     benedict,
     logger,
+    warn,
     pytest,
     json,
 
diff --git a/ocp_resources/data_source.py b/ocp_resources/data_source.py
index 247bca42da..5f936d1ff3 100644
--- a/ocp_resources/data_source.py
+++ b/ocp_resources/data_source.py
@@ -1,46 +1,38 @@
+from warnings import warn
 from kubernetes.dynamic.exceptions import ResourceNotFoundError
-
-from ocp_resources.constants import TIMEOUT_4MINUTES
 from ocp_resources.persistent_volume_claim import PersistentVolumeClaim
 from ocp_resources.resource import NamespacedResource
+from ocp_resources.volume_snapshot import VolumeSnapshot
 
 
 class DataSource(NamespacedResource):
+    """
+    DataSource object.
+
+    https://kubevirt.io/cdi-api-reference/main/definitions.html#_v1beta1_datasource
+    """
+
     api_group = NamespacedResource.ApiGroup.CDI_KUBEVIRT_IO
 
-    def __init__(
-        self,
-        name=None,
-        namespace=None,
-        client=None,
-        source=None,
-        teardown=True,
-        yaml_file=None,
-        delete_timeout=TIMEOUT_4MINUTES,
-        **kwargs,
-    ):
-        super().__init__(
-            name=name,
-            namespace=namespace,
-            client=client,
-            teardown=teardown,
-            yaml_file=yaml_file,
-            delete_timeout=delete_timeout,
-            **kwargs,
-        )
-        self.source = source
+    def __init__(self, source=None, **kwargs):
+        """
+        Args:
+            source (dict): The source of the data.
+        """
+        super().__init__(**kwargs)
+        self._source = source
 
     def to_dict(self):
         super().to_dict()
         if not self.yaml_file:
-            self.res.update({
-                "spec": {
-                    "source": self.source,
-                },
-            })
+            if not self._source:
+                raise ValueError("Passing yaml_file or parameter 'source' is required")
+
+            self.res["spec"]["source"] = self._source
 
     @property
     def pvc(self):
+        warn("pvc will be deprecated in v4.16, Use source instead", DeprecationWarning, stacklevel=2)
         data_source_pvc = self.instance.spec.source.pvc
         pvc_name = data_source_pvc.name
         pvc_namespace = data_source_pvc.namespace
@@ -55,3 +47,15 @@ def pvc(self):
                 f"dataSource {self.name} is pointing to a non-existing PVC, name:"
                 f" {pvc_name}, namespace: {pvc_namespace}"
             )
+
+    @property
+    def source(self):
+        _instance_source = self.instance.spec.source
+        _source = [*_instance_source][0][0]
+        _source_mapping = {"pvc": PersistentVolumeClaim, "snapshot": VolumeSnapshot}
+
+        return _source_mapping[_source](
+            client=self.client,
+            name=_instance_source[_source].name,
+            namespace=_instance_source[_source].namespace,
+        )