Pull-through streaming

pulp_rust/app/serializers.py

@@ -1,12 +1,33 @@
-from gettext import gettext as _
+import logging
 
+from gettext import gettext as _
+from django.db.utils import IntegrityError
+from pydantic import TypeAdapter, ValidationError
 from rest_framework import serializers
 
 from pulpcore.plugin import models as core_models
 from pulpcore.plugin import serializers as core_serializers
+from pulpcore.plugin.util import get_domain
 
 from . import models
 
+log = logging.getLogger(__name__)
+
+
+class IndexRootSerializer(serializers.Serializer):
+    """
+    A Serializer for summary information of an index.
+    """
+
+    dl = serializers.CharField(help_text=_("URL of the index root"), read_only=True)
+    api = serializers.CharField(help_text=_("URL of the API root"), read_only=True)
+    auth_required = serializers.BooleanField(
+        help_text=_(
+            "Indicates whether this is a private registry that requires all operations to be authenticated"
+        ),
+        read_only=True,
+    )
+
 
 class RustDependencySerializer(serializers.ModelSerializer):
     """
@@ -178,31 +199,187 @@ class Meta:
         model = models.RustContent
 
 
-class RustRemoteSerializer(core_serializers.RemoteSerializer):
+class RustDependencySerializer(serializers.ModelSerializer):
     """
-    A Serializer for RustRemote.
+    Serializer for RustDependency.
 
-    Add any new fields if defined on RustRemote.
-    Similar to the example above, in RustContentSerializer.
-    Additional validators can be added to the parent validators list
+    Represents a single dependency entry from the Cargo package index.
+    """
 
-    For example::
+    name = serializers.CharField(
+        help_text=_("Dependency name as used in code (may be renamed via 'package' field)")
+    )
+
+    req = serializers.CharField(
+        help_text=_("Version requirement string (e.g., '^1.0', '>=0.2.3,<0.3')")
+    )
+
+    features = serializers.ListField(
+        child=serializers.CharField(),
+        default=list,
+        required=False,
+        help_text=_("List of feature flags to enable for this dependency"),
+    )
+
+    optional = serializers.BooleanField(
+        default=False, required=False, help_text=_("Whether this is an optional dependency")
+    )
+
+    default_features = serializers.BooleanField(
+        default=True,
+        required=False,
+        help_text=_("Whether to enable the dependency's default features"),
+    )
+
+    target = serializers.CharField(
+        allow_null=True,
+        required=False,
+        help_text=_("Platform-specific target (e.g., 'cfg(unix)', 'cfg(windows)')"),
+    )
+
+    kind = serializers.ChoiceField(
+        choices=[("normal", "Normal"), ("dev", "Development"), ("build", "Build")],
+        default="normal",
+        required=False,
+        help_text=_(
+            "Dependency type: 'normal' (runtime), 'dev' (development), or 'build' (build script)"
+        ),
+    )
+
+    registry = serializers.CharField(
+        allow_null=True,
+        required=False,
+        help_text=_("Alternative registry URL if dependency is from a different registry"),
+    )
+
+    package = serializers.CharField(
+        allow_null=True,
+        required=False,
+        help_text=_("Original crate name if the dependency was renamed"),
+    )
 
     class Meta:
-        validators = core_serializers.RemoteSerializer.Meta.validators
-            + [myValidator1, myValidator2]
+        model = models.RustDependency
+        fields = (
+            "name",
+            "req",
+            "features",
+            "optional",
+            "default_features",
+            "target",
+            "kind",
+            "registry",
+            "package",
+        )
+
+
+class RustContentSerializer(core_serializers.SingleArtifactContentSerializer):
+    """
+    Serializer for RustContent (Cargo package version).
+
+    Represents a single version of a Rust crate as defined in the Cargo registry
+    index specification. Includes package metadata, dependencies, and features.
+    """
+
+    name = serializers.CharField(help_text=_("Package name (crate name)"))
+
+    vers = serializers.CharField(help_text=_("Semantic version string (SemVer 2.0.0)"))
 
-    By default the 'policy' field in core_serializers.RemoteSerializer only validates the choice
-    'immediate'. To add on-demand support for more 'policy' options, e.g. 'streamed' or
-    'on_demand', re-define the 'policy' option as follows::
+    dependencies = RustDependencySerializer(
+        many=True, required=False, help_text=_("List of dependencies for this package version")
+    )
+
+    cksum = serializers.CharField(help_text=_("SHA256 checksum of the .crate file (tarball)"))
+
+    features = serializers.JSONField(
+        default=dict,
+        required=False,
+        help_text=_(
+            "Feature flags mapping - maps feature names to lists of features/dependencies "
+            "they enable"
+        ),
+    )
+
+    features2 = serializers.JSONField(
+        default=dict,
+        required=False,
+        allow_null=True,
+        help_text=_("Extended feature syntax support (newer registry format)"),
+    )
+
+    yanked = serializers.BooleanField(
+        default=False,
+        required=False,
+        help_text=_("Whether this version has been yanked (removed from normal use)"),
+    )
+
+    links = serializers.CharField(
+        allow_null=True,
+        required=False,
+        help_text=_("Name of native library this package links to (from Cargo.toml 'links' field)"),
+    )
+
+    v = serializers.IntegerField(
+        default=1, required=False, help_text=_("Schema version of the index entry format")
+    )
+    rust_version = serializers.CharField(
+        allow_null=True,
+        required=False,
+        help_text=_("Minimum Rust compiler version required (MSRV)"),
+    )
+
+    def create(self, validated_data):
+        """Create RustContent and related dependencies."""
+        dependencies_data = validated_data.pop("dependencies", [])
+        content = super().create(validated_data)
+
+        # Create dependency records
+        for dep_data in dependencies_data:
+            models.RustDependency.objects.create(content=content, **dep_data)
+
+        return content
+
+    def update(self, instance, validated_data):
+        """Update RustContent and related dependencies."""
+        dependencies_data = validated_data.pop("dependencies", None)
+
+        instance = super().update(instance, validated_data)
+
+        if dependencies_data is not None:
+            # Replace all dependencies
+            instance.dependencies.all().delete()
+            for dep_data in dependencies_data:
+                models.RustDependency.objects.create(content=instance, **dep_data)
+
+        return instance
+
+    class Meta:
+        fields = core_serializers.SingleArtifactContentSerializer.Meta.fields + (
+            "name",
+            "vers",
+            "dependencies",
+            "cksum",
+            "features",
+            "features2",
+            "yanked",
+            "links",
+            "v",
+            "rust_version",
+        )
+        model = models.RustContent
+
+
+class RustRemoteSerializer(core_serializers.RemoteSerializer):
+    """
+    A Serializer for RustRemote.
+    """
 
     policy = serializers.ChoiceField(
         help_text="The policy to use when downloading content. The possible values include: "
-                  "'immediate', 'on_demand', and 'streamed'. 'immediate' is the default.",
+        "'immediate', 'on_demand', and 'streamed'. 'streamed' is the default.",
         choices=models.Remote.POLICY_CHOICES,
-        default=models.Remote.IMMEDIATE
+        default=models.Remote.STREAMED,
     )
-    """
 
     class Meta:
         fields = core_serializers.RemoteSerializer.Meta.fields
@@ -258,3 +435,31 @@ class Meta:
     class Meta:
         fields = core_serializers.DistributionSerializer.Meta.fields + ("allow_uploads", "remote")
         model = models.RustDistribution
+
+
+class RepositoryAddCachedContentSerializer(
+    core_serializers.ValidateFieldsMixin, serializers.Serializer
+):
+    remote = core_serializers.DetailRelatedField(
+        required=False,
+        view_name_pattern=r"remotes(-.*/.*)-detail",
+        queryset=models.Remote.objects.all(),
+        help_text=_(
+            "A remote to use to identify content that was cached. This will override a "
+            "remote set on repository."
+        ),
+    )
+
+    def validate(self, data):
+        data = super().validate(data)
+        repository = None
+        if "repository_pk" in self.context:
+            repository = models.Repository.objects.get(pk=self.context["repository_pk"])
+        remote = data.get("remote", None) or getattr(repository, "remote", None)
+
+        if not remote:
+            raise serializers.ValidationError(
+                {"remote": _("This field is required since a remote is not set on the repository.")}
+            )
+        self.check_cross_domains({"repository": repository, "remote": remote})
+        return data

pulp_rust/app/settings.py

@@ -1,6 +1,3 @@
-"""
-Check `Plugin Writer's Guide`_ for more details.
+import socket
 
-.. _Plugin Writer's Guide:
-    https://pulpproject.org/pulpcore/docs/dev/
-"""
+CRATES_IO_API_HOSTNAME = "https://" + socket.getfqdn()

pulp_rust/app/tasks/__init__.py

@@ -1 +1,2 @@
 from .synchronizing import synchronize  # noqa
+from .streaming import add_cached_content_to_repository  # noqa

pulp_rust/app/tasks/streaming.py

@@ -0,0 +1,40 @@
+import datetime
+
+from asgiref.sync import sync_to_async
+
+from pulpcore.plugin.models import Content, ContentArtifact, RemoteArtifact
+from pulpcore.plugin.tasking import add_and_remove
+
+from pulp_rust.app.models import RustRemote, RustRepository
+
+
+async def aadd_and_remove(*args, **kwargs):
+    return await sync_to_async(add_and_remove)(*args, **kwargs)
+
+
+def add_cached_content_to_repository(repository_pk=None, remote_pk=None):
+    """
+    Create a new repository version by adding content that was cached by pulpcore-content when
+    streaming it from a remote.
+
+    Args:
+        repository_pk (uuid): The primary key for a Repository for which a new Repository Version
+            should be created.
+        remote_pk (uuid): The primary key for a Remote which will be used to identify Content
+            created by pulpcore-content when it streamed it to clients.
+    """
+    repository = RustRepository.objects.get(pk=repository_pk)
+    remote = RustRemote.objects.get(pk=remote_pk)
+
+    latest_version = repository.latest_version()
+
+    if latest_version.number == 0:
+        date_min = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+    else:
+        date_min = latest_version.pulp_created
+    with repository.new_version(base_version=None) as new_version:
+        ca_id_list = RemoteArtifact.objects.filter(
+            remote=remote, pulp_created__gte=date_min
+        ).values_list("content_artifact")
+        content_list = ContentArtifact.objects.filter(pk__in=ca_id_list).values_list("content")
+        new_version.add_content(Content.objects.filter(pk__in=content_list))

pulp_rust/app/tasks/synchronizing.py

@@ -73,6 +73,8 @@ async def run(self):
         downloader = self.remote.get_downloader(url=self.remote.url)
         result = await downloader.run()
         # Use ProgressReport to report progress
+        raise NotImplemented("Not implemented")
+
         for entry in self.read_my_metadata_file_somehow(result.path):
             unit = RustContent(entry)  # make the content unit in memory-only
             artifact = Artifact(entry)  # make Artifact in memory-only

pulp_rust/app/urls.py

@@ -0,0 +1,17 @@
+from django.conf import settings
+from django.urls import path, re_path
+
+from pulp_rust.app.views import IndexRoot, CargoApiViewSet
+
+if settings.DOMAIN_ENABLED:
+    path_re = r"(?P<pulp_domain>[-a-zA-Z0-9_]+)/(?P<name>[\w-]+)/(?P<path>.*)"
+    CRATES_IO_URL = "<slug:pulp_domain>/<path:path>/"
+else:
+    path_re = r"(?P<name>[\w-]+)/(?P<path>.*)"
+    CRATES_IO_URL = "<path:path>/"
+
+
+urlpatterns = [
+    path(CRATES_IO_URL, IndexRoot.as_view({"get": "retrieve"}), name="crates-io-root"),
+    re_path(rf"^pulp/cargo/{path_re}$", CargoApiViewSet.as_view({"get": "retrieve"})),
+]