L1L`XPL0IT

✏️ Edit: sync.py

📁 /snap/google-cloud-cli/current/lib/surface/run/services/dev/sync.py

# -*- coding: utf-8 -*- #
# Copyright 2026 Google LLC. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Command to Sync local workspace to a Cloud Run Service."""

import argparse

from googlecloudsdk.api_lib.run import constants
from googlecloudsdk.api_lib.run import service
from googlecloudsdk.calliope import base
from googlecloudsdk.command_lib.run import config_changes
from googlecloudsdk.command_lib.run import connection_context
from googlecloudsdk.command_lib.run import container_parser
from googlecloudsdk.command_lib.run import deletion
from googlecloudsdk.command_lib.run import exceptions
from googlecloudsdk.command_lib.run import flags
from googlecloudsdk.command_lib.run import pretty_print
from googlecloudsdk.command_lib.run import resource_args
from googlecloudsdk.command_lib.run import serverless_operations
from googlecloudsdk.command_lib.run.sourcedeploys import deploy_util
from googlecloudsdk.command_lib.run.sync import sync_util
from googlecloudsdk.command_lib.util.args import map_util
from googlecloudsdk.command_lib.util.concepts import concept_parsers
from googlecloudsdk.command_lib.util.concepts import presentation_specs
from googlecloudsdk.core import execution_utils
from googlecloudsdk.core import log
from googlecloudsdk.core.console import console_io

_MIN_INSTANCES_ANNOTATION = 'autoscaling.knative.dev/minScale'
_MAX_INSTANCES_ANNOTATION = 'autoscaling.knative.dev/maxScale'

def ContainerArgGroup(release_track=base.ReleaseTrack.GA):
  """Returns an argument group with all container deploy & sync args."""

help_text = """
Container Flags

The following flags apply to the container.
"""
  group = base.ArgumentGroup(help=help_text)
  group.AddArgument(flags.SourceArg())
  group.AddArgument(flags.PortArg())
  group.AddArgument(flags.MutexBuildEnvVarsFlags())
  group.AddArgument(flags.MutexEnvVarsFlags(release_track=release_track))
  group.AddArgument(flags.MemoryFlag())
  group.AddArgument(flags.CpuFlag())
  group.AddArgument(flags.ArgsFlag())
  group.AddArgument(flags.SecretsFlags())
  group.AddArgument(flags.AddCommandAndFunctionFlag())
  group.AddArgument(flags.DependsOnFlag())
  group.AddArgument(flags.AddVolumeMountFlag())
  group.AddArgument(flags.RemoveVolumeMountFlag())
  group.AddArgument(flags.ClearVolumeMountsFlag())
  group.AddArgument(flags.StartupProbeFlag())
  group.AddArgument(flags.SandboxLauncherFlag())
  group.AddArgument(flags.Http2Flag())
  group.AddArgument(flags.WorkdirFlag())
  group.AddArgument(flags.BuildServiceAccountMutexGroup())
  group.AddArgument(flags.LivenessProbeFlag())
  group.AddArgument(flags.ReadinessProbeFlag())
  group.AddArgument(flags.GpuFlag())
  return group

def NecessaryChangesForServicesDevSync(
    args: argparse.Namespace,
) -> list[config_changes.ConfigChanger]:
  """Adds necessary changes for service used by dev sync, if not specified by flags.

1. Set SSH Enabled for the new Service
  2. Set Execution Environment to gen2
  3. Set min and max number of instances to 1

Args:
    args: The command-line arguments.

Returns:
    A list of Config Changes objects to apply to the Service for Dev Sync.
  """
  changes = [
      config_changes.SetAnnotationChange(
          service.SERVICE_SSH_ENABLED_ANNOTATION, 'true'
      ),
  ]
  if getattr(args, 'execution_environment', None) is None:
    changes.append(config_changes.SandboxChange(sandbox='gen2'))

if getattr(args, 'min', None) is None:
    changes.append(
        config_changes.SetTemplateAnnotationChange(
            _MIN_INSTANCES_ANNOTATION, '1'
        )
    )
  if getattr(args, 'max', None) is None:
    changes.append(
        config_changes.SetTemplateAnnotationChange(
            _MAX_INSTANCES_ANNOTATION, '1'
        )
    )

return changes

@base.ReleaseTracks(base.ReleaseTrack.ALPHA)
@base.DefaultUniverseOnly
@base.RegionalEndpointsSupported
@base.Hidden
class Sync(base.Command):
  """A command to sync local workspace to a Cloud Run Service."""

detailed_help = {
      'DESCRIPTION': (
          """\
          Syncs the local workspace to Cloud Run Service.
          """
      ),
      'EXAMPLES': (
          """\
          To sync the local workspace to a Cloud Run Service:

$ {command} my-service --source=.
          """
      ),
  }

@classmethod
  def CommonArgs(cls, parser):
    flags.DevSyncCleanupFlag(parser)
    parser.add_argument(
        '--iap-tunnel-url-override',
        hidden=True,
        help=(
            'Allows for overriding the connection endpoint for integration'
            ' testing.'
        ),
    )

flags.AddAllowUnauthenticatedFlag(parser)
    flags.AddAllowUnencryptedBuildFlag(parser)
    flags.AddBinAuthzPolicyFlags(parser)
    flags.AddBinAuthzBreakglassFlag(parser)
    flags.AddCloudSQLFlags(parser)
    flags.AddCmekKeyFlag(parser)
    flags.AddCmekKeyRevocationActionTypeFlag(parser)
    flags.AddCpuThrottlingFlag(parser)
    flags.AddCustomAudiencesFlag(parser)
    flags.AddDefaultUrlFlag(parser)
    flags.AddDeployHealthCheckFlag(parser)
    flags.AddDescriptionFlag(parser)
    flags.AddEgressSettingsFlag(parser)
    flags.AddEncryptionKeyShutdownHoursFlag(parser)
    flags.AddGpuTypeFlag(parser)
    flags.GpuZonalRedundancyFlag(parser)
    flags.AddRevisionSuffixArg(parser)
    flags.AddSandboxArg(parser)
    flags.AddSessionAffinityFlag(parser)
    flags.AddStartupCpuBoostFlag(parser)
    flags.AddVpcConnectorArgs(parser)
    flags.AddVpcNetworkGroupFlagsForUpdate(parser)
    flags.RemoveContainersFlag().AddToParser(parser)
    flags.AddVolumesFlags(parser, cls.ReleaseTrack())
    flags.AddServiceMinMaxInstancesFlag(parser)
    flags.AddInvokerIamCheckFlag(parser)
    flags.AddScalingFlag(parser, release_track=cls.ReleaseTrack())

# Flags specific to connecting to a cluster
    flags.AddEndpointVisibilityEnum(parser)
    flags.CONFIG_MAP_FLAGS.AddToParser(parser)

# Flags not specific to any platform
    service_presentation = presentation_specs.ResourcePresentationSpec(
        'SERVICE',
        resource_args.GetServiceResourceSpec(prompt=True),
        'Service to sync to.',
        required=True,
        prefixes=False,
    )
    if cls.ReleaseTrack() != base.ReleaseTrack.GA:
      flags.AddDryRunFlag(parser)
    flags.AddConcurrencyFlag(parser)
    flags.AddTimeoutFlag(parser)
    flags.AddAsyncFlag(parser)
    flags.AddLabelsFlags(parser)
    flags.AddGeneralAnnotationFlags(parser)
    flags.AddMinInstancesFlag(parser)
    flags.AddMaxInstancesFlag(parser)
    flags.AddNoTrafficFlag(parser)
    flags.AddDeployTagFlag(parser)
    flags.AddServiceAccountFlag(parser)
    flags.AddClientNameAndVersionFlags(parser)
    flags.AddIngressFlag(parser)
    flags.AddRegionsArg(parser)
    flags.AddIapFlag(parser)
    concept_parsers.ConceptParser([service_presentation]).AddToParser(parser)

@classmethod
  def Args(cls, parser):
    cls.CommonArgs(parser)
    container_args = ContainerArgGroup(cls.ReleaseTrack())
    container_parser.AddContainerFlags(
        parser, container_args, cls.ReleaseTrack()
    )

def _CleanupService(self, args, service_ref):
    pretty_print.Info('Waiting for service to be deleted...')
    conn_context = connection_context.GetConnectionContext(
        args, flags.Product.RUN, self.ReleaseTrack()
    )
    with serverless_operations.Connect(conn_context) as client:
      deletion.Delete(
          service_ref, client.GetService, client.DeleteService, False
      )
      log.DeletedResource(service_ref.servicesId, 'service')

def _ValidateServiceForDevSync(self, svc: service.Service):
    """Validates that the service is valid for dev sync."""
    min_scale = svc.annotations.get(_MIN_INSTANCES_ANNOTATION)
    max_scale = svc.annotations.get(_MAX_INSTANCES_ANNOTATION)

if max_scale != '1' or min_scale != '1':
      pretty_print.Info(
          '{bold}**[Warning]**{reset} Service is not restricted to single'
          ' instance. Dev Sync will only update one instance. Set min & max'
          ' instances to 1 to avoid routing traffic to un-synced instances.'
      )

def _GetService(self, args, service_ref) -> service.Service:
    conn_context = connection_context.GetConnectionContext(
        args, flags.Product.RUN, self.ReleaseTrack()
    )
    with serverless_operations.Connect(conn_context) as client:
      return client.GetService(service_ref)

def Run(self, args):
    flags.ValidatePublicFlags(args)

service_ref = args.CONCEPTS.service.Parse()
    flags.ValidateResource(service_ref)
    args.release_track = self.ReleaseTrack()
    args.project = flags.GetProjectID(args)

args.region = flags.GetRegion(args, prompt=False)
    if not args.region:
      raise exceptions.ArgumentError(
          'Missing required argument [region]. Set --region flag or set'
          ' run/region property.'
      )

if not args.source:
      if console_io.CanPrompt():
        args.source = flags.PromptForDefaultSource()
      else:
        raise exceptions.ArgumentError(
            'The --source flag must be provided to specify the source for dev'
            ' sync.'
        )

build_env_var_flags = map_util.GetMapFlagsFromArgs('build-env-vars', args)
    args.build_env_vars = (
        map_util.ApplyMapFlags(None, **build_env_var_flags) or {}
    )
    args.build_env_vars['GOOGLE_DEVSYNC'] = 'true'

args.deployment_name = service_ref.Name()
    svc = self._GetService(args, service_ref)
    service_exists = svc is not None

if service_exists:
      if args.cleanup:
        raise exceptions.ArgumentError(
            'The --cleanup flag is not supported for dev sync to an existing'
            ' Service.'
        )
      self._ValidateServiceForDevSync(svc)

pretty_print.Info(f'Syncing to existing service: {svc.name}')
      url = svc.latest_url or (svc.status.url if svc.status else '')
      if url:
        pretty_print.Info(f'Service URL: {{bold}}{url}{{reset}}')
    else:
      changes = NecessaryChangesForServicesDevSync(args)
      deploy_util.DeployServiceFromSource(
          service_ref=service_ref,
          source=args.source,
          region=args.region,
          args=args,
          release_track=self.ReleaseTrack(),
          changes=changes,
      )

try:
      with execution_utils.RaisesKeyboardInterrupt():
        sync_util.Sync(
            args=args,
            workload_type=constants.WorkloadType.SERVICE,
            source=args.source,
        ).Run()
    except KeyboardInterrupt:
      pretty_print.Info(
          'Received Keyboard Interrupt... Dev Sync Session terminated'
      )
    finally:
      if args.cleanup and not service_exists:
        self._CleanupService(args, service_ref)

Cancel

💻 Terminal

📤 Upload

📁 /snap/google-cloud-cli/current/lib/surface/run/services/dev

//snap/google-cloud-cli/current/lib/surface/run/services/dev

Name	Size	Perm	Modified	Actions
📁 __pycache__/	-	0775	2026-09-29 12:17	🗑 Delete
__init__.py	1.1K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
sync.py	10.9K	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del
sync.yaml	82B	0644	1980-01-01 08:00	👁 View ✏️ Edit 📝 Rename 📦 Move 🗑 Del