feat: implement Target that support read and write

Signed-off-by: jiwangCHEN <hyper1char@gmail.com>

Author: jiwangCHEN

oras/copy/__init__.py

@@ -26,7 +26,7 @@
 
 Adapters:
     RegistryTarget      - Adapts Registry + container to Target/ReferencePusher/Mounter
-    LayoutTarget        - Adapts Layout directory to ReadOnlyTarget
+    LayoutTarget        - Adapts Layout directory to Target (full read/write)
 """
 
 __author__ = "The ORAS Authors"

oras/copy/adapters.py

@@ -7,7 +7,7 @@
 RegistryTarget wraps a Registry + container string into a Target +
 ReferencePusher + Mounter scoped to a single repository.
 
-LayoutTarget wraps a Layout directory into a ReadOnlyTarget.
+LayoutTarget wraps a Layout directory into a full read/write Target.
 """
 
 __author__ = "The ORAS Authors"
@@ -17,16 +17,23 @@
 import os
 import pathlib
 import re
+import shutil
 import tempfile
-from typing import BinaryIO, Callable
+import threading
+from typing import TYPE_CHECKING, BinaryIO, Callable, Optional
 
 import oras.defaults
 import oras.utils
 from oras.copy.descriptor import is_manifest
 from oras.layout.layout import Layout
 from oras.provider import Registry
 from oras.types import Descriptor
-from oras.utils.fileio import read_json
+from oras.utils.fileio import read_json, write_json
+
+if TYPE_CHECKING:
+    import requests
+
+    from oras.copy.options import CopyOptions
 
 _VALID_DIGEST_RE = re.compile(r"^[a-z0-9]+:[a-f0-9]+$")
 
@@ -50,10 +57,20 @@ class RegistryTarget:
     by the container string passed at construction time.
     """
 
-    def __init__(self, registry: Registry, container: str):
+    def __init__(
+        self,
+        registry: Registry,
+        container: str,
+        opts: "Optional[CopyOptions]" = None,
+    ):
         self._registry = registry
         self._container = registry.get_container(container)
         self._registry.auth.load_configs(self._container)
+        self._opts = opts
+        # Records the most recent manifest PUT response (from push /
+        # push_reference) so callers like Registry.push can return the real
+        # upload response rather than issuing a follow-up GET.
+        self.last_manifest_response: "Optional[requests.Response]" = None
 
     def _manifest_url(self, ref: str) -> str:
         """Build full manifest URL for a reference (tag or digest)."""
@@ -95,15 +112,31 @@ def push(self, desc: Descriptor, content: BinaryIO) -> None:
             response = self._registry.do_request(
                 url, "PUT", data=data, headers=headers
             )
+            self.last_manifest_response = response
             self._registry._check_200_response(response)
         else:
-            data = content.read()
             tmp = None
             try:
                 tmp = tempfile.NamedTemporaryFile(delete=False)
-                tmp.write(data)
+                # Stream rather than buffer: large layers (and chunked
+                # uploads) must not be fully materialized in memory.
+                shutil.copyfileobj(content, tmp)
                 tmp.close()
-                self._registry.upload_blob(tmp.name, self._container, desc)
+                do_chunked = False
+                chunk_size = oras.defaults.default_chunksize
+                if self._opts is not None:
+                    do_chunked = self._opts.graph.do_chunked
+                    chunk_size = (
+                        self._opts.graph.chunk_size
+                        or oras.defaults.default_chunksize
+                    )
+                self._registry.upload_blob(
+                    tmp.name,
+                    self._container,
+                    desc,
+                    do_chunked=do_chunked,
+                    chunk_size=chunk_size,
+                )
             finally:
                 if tmp is not None:
                     try:
@@ -144,6 +177,7 @@ def push_reference(
         response = self._registry.do_request(
             url, "PUT", data=data, headers=headers
         )
+        self.last_manifest_response = response
         self._registry._check_200_response(response)
 
     def mount(
@@ -189,20 +223,47 @@ def mount(
 
 class LayoutTarget:
     """
-    Adapts a Layout directory into the copy engine's ReadOnlyTarget protocol.
+    Adapts a Layout directory into the copy engine's Target protocol.
 
-    Provides read-only access to the OCI layout's content-addressable blobs
-    and resolves references via the layout's index.json annotations.
+    Provides full read/write access to the OCI layout's content-addressable
+    blobs and manages references via the layout's index.json annotations.
+
+    Implements Target (fetch, exists, push, tag, resolve), so it can serve
+    as both source and destination in :func:`oras.copy.copy`.
     """
 
     def __init__(self, layout: Layout):
         self._layout = layout
+        self._index_lock = threading.Lock()
 
     @staticmethod
     def _validate_digest(digest: str) -> None:
         if not _VALID_DIGEST_RE.match(digest):
             raise ValueError(f"invalid digest format: {digest!r}")
 
+    def _ensure_initialized(self) -> None:
+        """Create OCI layout structure (oci-layout, index.json, blobs/) if missing."""
+        layout_dir = pathlib.Path(self._layout._oci_layout_path)
+        layout_dir.mkdir(parents=True, exist_ok=True)
+        (layout_dir / oras.defaults.oci_blobs_dir).mkdir(exist_ok=True)
+
+        oci_layout_path = layout_dir / oras.defaults.oci_layout_file
+        if not oci_layout_path.exists():
+            write_json(
+                {"imageLayoutVersion": oras.defaults.oci_layout_version_pin},
+                str(oci_layout_path),
+            )
+
+        index_path = layout_dir / oras.defaults.oci_image_index_file
+        if not index_path.exists():
+            write_json(
+                {
+                    "schemaVersion": oras.defaults.oci_index_schema_version,
+                    "manifests": [],
+                },
+                str(index_path),
+            )
+
     def fetch(self, desc: Descriptor) -> BinaryIO:
         """Fetch blob content by digest, returning an open file handle."""
         digest = desc["digest"]
@@ -217,6 +278,75 @@ def exists(self, desc: Descriptor) -> bool:
         path = self._layout.digest_to_blob_path(digest)
         return path.exists()
 
+    def push(self, desc: Descriptor, content: BinaryIO) -> None:
+        """Write blob content to the layout (content-addressed, deduplicated).
+
+        Matching oras-go's Store.Push: skips silently if the blob already
+        exists (same digest), otherwise writes atomically via a temp file +
+        rename so concurrent writers never observe a partial blob.
+        """
+        digest = desc["digest"]
+        self._validate_digest(digest)
+
+        self._ensure_initialized()
+
+        path = self._layout.digest_to_blob_path(digest)
+        if path.exists():
+            return  # already present — deduplicate silently
+
+        path.parent.mkdir(parents=True, exist_ok=True)
+
+        tmp_fd, tmp_name = tempfile.mkstemp(dir=str(path.parent))
+        try:
+            # Stream rather than buffer: large blobs must not be fully
+            # materialized in memory.
+            with os.fdopen(tmp_fd, "wb") as f:
+                shutil.copyfileobj(content, f)
+            os.replace(tmp_name, str(path))
+        except Exception:
+            try:
+                os.unlink(tmp_name)
+            except OSError:
+                pass
+            raise
+
+    def tag(self, desc: Descriptor, reference: str) -> None:
+        """Update index.json to associate the descriptor with a reference tag.
+
+        Matching oras-go's Store.Tag: finds an existing index.json entry for
+        this reference and replaces it, or appends a new one.
+        Thread-safe via _index_lock.
+        """
+        self._ensure_initialized()
+
+        layout_dir = pathlib.Path(self._layout._oci_layout_path)
+        index_path = layout_dir / oras.defaults.oci_image_index_file
+
+        new_entry = {
+            "mediaType": desc.get("mediaType", ""),
+            "digest": desc["digest"],
+            "size": desc.get("size", 0),
+            "annotations": {oras.defaults.oci_ref_name_annotation: reference},
+        }
+
+        with self._index_lock:
+            index_data = read_json(str(index_path))
+            manifests = index_data.setdefault("manifests", [])
+
+            for i, entry in enumerate(manifests):
+                if (
+                    entry.get("annotations", {}).get(
+                        oras.defaults.oci_ref_name_annotation
+                    )
+                    == reference
+                ):
+                    manifests[i] = new_entry
+                    break
+            else:
+                manifests.append(new_entry)
+
+            write_json(index_data, str(index_path))
+
     def resolve(self, reference: str) -> Descriptor:
         """Resolve a reference tag to a descriptor via index.json."""
         layout_dir = pathlib.Path(self._layout._oci_layout_path)

oras/copy/copy.py

@@ -246,14 +246,16 @@ def _copy_cached_node_with_reference(
     reference: str,
 ) -> None:
     """
-    Push a cached node to the destination with a reference tag.
-
-    The content must already be in the proxy cache (as is the case
-    for manifests/indexes that were fetched during graph traversal).
-
-    Matches oras-go's copyCachedNodeWithReference.
+    Push a node to the destination with a reference tag.
+
+    The content is normally already in the proxy cache (fetched during
+    graph traversal). When the root already exists at the destination its
+    sub-DAG is skipped and never fetched, so the cache may miss; in that
+    case we fall back to fetching from the source via the proxy. This
+    matches oras-go's copyCachedNodeWithReference, which uses FetchCached
+    (cache-or-source), not a cache-only read.
     """
-    rc = proxy.cache.fetch(desc)
+    rc = proxy.fetch(desc)
     try:
         data = rc.read()
     finally:

oras/copy/options.py

@@ -42,6 +42,10 @@ class CopyGraphOptions:
         on_mounted: Called when a blob is successfully mounted.
         find_successors: Custom function to discover child nodes of
             a descriptor. If None, content.successors is used.
+        do_chunked: If True, blob uploads to a registry destination use
+            chunked upload. Honored by RegistryTarget.push.
+        chunk_size: Chunk size in bytes for chunked uploads. If <= 0,
+            defaults to oras.defaults.default_chunksize.
     """
 
     concurrency: int = 0
@@ -52,6 +56,8 @@ class CopyGraphOptions:
     mount_from: Optional[Callable[[Descriptor], List[str]]] = None
     on_mounted: Optional[Callable[[Descriptor], None]] = None
     find_successors: Optional[Callable] = None
+    do_chunked: bool = False
+    chunk_size: int = 0
 
 
 @dataclass

oras/layout/layout.py

@@ -524,12 +524,12 @@ def push_to_registry(
 
     def as_target(self) -> LayoutTarget:
         """
-        Return a copy-engine ReadOnlyTarget adapter for this layout.
+        Return a copy-engine Target adapter for this layout.
 
         The returned LayoutTarget can be passed directly to
-        :func:`oras.copy.copy` as a source.
+        :func:`oras.copy.copy` as either source or destination.
 
-        :return: a ReadOnlyTarget adapter wrapping this layout
+        :return: a Target adapter wrapping this layout
         :rtype: oras.copy.adapters.LayoutTarget
         """
         from oras.copy.adapters import LayoutTarget
@@ -552,7 +552,10 @@ def copy_to_registry(
 
         :param provider: Registry provider instance
         :type provider: oras.provider.Registry
-        :param target: target registry/repository (e.g., "ghcr.io/user/repo:v1.0")
+        :param target: target registry/repository with the destination tag
+            (e.g., "ghcr.io/user/repo:v1.0"). The tag/digest in this URL
+            determines the destination reference at the registry. When no
+            tag is present, "latest" is used.
         :type target: str
         :param tag: source tag to read from the layout's index.json annotations (default: "latest")
         :type tag: str
@@ -563,12 +566,52 @@ def copy_to_registry(
         :raises FileNotFoundError: if layout or blobs don't exist
         :raises ValueError: if layout is invalid or tag not found
         """
+        from oras.container import Container
         from oras.copy import copy as copy_fn
         from oras.copy.adapters import LayoutTarget, RegistryTarget
 
         src = LayoutTarget(self)
-        dst = RegistryTarget(provider, target)
-        return copy_fn(src, tag, dst, tag, opts)
+        dst = RegistryTarget(provider, target, opts)
+        dst_container = Container(target)
+        dst_ref = dst_container.digest or dst_container.tag
+        return copy_fn(src, tag, dst, dst_ref, opts)
+
+    def copy_from_registry(
+        self,
+        provider: Registry,
+        source: str,
+        tag: str = "latest",
+        opts: CopyOptions = None,
+    ) -> Descriptor:
+        """
+        Copy content from a remote registry into this layout using the copy engine.
+
+        Uses the copy engine's DAG-aware graph traversal to pull all content
+        from the source registry into this layout directory.
+        This is the recommended way to pull a registry artifact into a layout.
+
+        :param provider: Registry provider instance
+        :type provider: oras.provider.Registry
+        :param source: source registry/repository (e.g., "ghcr.io/user/repo:v1.0")
+        :type source: str
+        :param tag: tag to write in the layout's index.json annotation (default: "latest")
+        :type tag: str
+        :param opts: copy options (default: None for default settings)
+        :type opts: oras.copy.options.CopyOptions
+        :return: the root descriptor that was copied
+        :rtype: dict
+        :raises FileNotFoundError: if source tag/digest is not found in registry
+        :raises ValueError: if source is invalid
+        """
+        from oras.container import Container
+        from oras.copy import copy as copy_fn
+        from oras.copy.adapters import LayoutTarget, RegistryTarget
+
+        src = RegistryTarget(provider, source, opts)
+        dst = LayoutTarget(self)
+        src_container = Container(source)
+        src_ref = src_container.digest or src_container.tag
+        return copy_fn(src, src_ref, dst, tag, opts)
 
     def pull_from_registry(
         self,