Managing files across local development environments and GCP Cloud Storage buckets is a daily chore for DevOps teams. Without automation, engineers end up
Managing files across local development environments and GCP Cloud Storage buckets is a daily chore for DevOps teams. Without automation, engineers end up with stale backups, orphaned objects, and inconsistent directory trees between environments. Worse, GCS buckets silently accumulate old artifacts (logs, exports, temp files) that inflate storage costs month over month — and cleaning them up manually is error-prone.
This script solves both problems in one tool: it performs a bidirectional-aware sync of a local directory against a GCS bucket (uploading new/changed files, optionally deleting remote files that no longer exist locally), and it applies a lifecycle policy that deletes remote objects older than a configurable age (90d, 24h, 30m). It supports glob-based excludes, dry-run mode, and parallel uploads so it works equally well for a 10-file config folder or a 100,000-object data lake.
The script uses the official google-cloud-storage client. On startup it parses CLI args, resolves the bucket name, and builds a SyncPlan by comparing local file metadata (path, size, mtime) against remote blob metadata (name, size, updated). Files are considered "in sync" if size matches and the remote updated timestamp is newer than the local mtime — this avoids re-uploading unchanged files and avoids relying on MD5 hashes (which GCS exposes but are unreliable for composite objects).
Deletion is handled in two phases: age-based (--delete-after 90d removes remote objects whose updated timestamp is older than the threshold) and mirror-based (--mirror removes remote objects with no local counterpart). Both honor --exclude globs and --dry-run. Uploads run through a ThreadPoolExecutor with a configurable worker count, since google-cloud-storage is thread-safe per-client. All destructive operations are logged to stderr and summarized at the end so operators can audit exactly what changed.
#!/usr/bin/env python3
"""
GCP Cloud Storage Sync and Lifecycle Manager
Syncs a local directory to a GCS bucket, optionally mirrors deletions,
and enforces an age-based lifecycle policy on remote objects.
Usage:
python3 gcs_sync.py --bucket gs://my-bucket --local ./data \
--delete-after 90d --exclude '*.tmp' --exclude '.git/*'
"""
from __future__ import annotations
import argparse
import fnmatch
import logging
import os
import re
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Iterable
from google.api_core import exceptions as gcp_exceptions
from google.cloud import storage
# ---------------------------------------------------------------------------
# Configuration (env vars with sane defaults)
# ---------------------------------------------------------------------------
DEFAULT_WORKERS = int(os.environ.get("GCS_SYNC_WORKERS", "8"))
DEFAULT_PREFIX = os.environ.get("GCS_SYNC_PREFIX", "")
DEFAULT_CHUNK_SIZE = int(os.environ.get("GCS_SYNC_CHUNK_MB", "8")) * 1024 * 1024
LOG_LEVEL = os.environ.get("GCS_SYNC_LOG_LEVEL", "INFO").upper()
logging.basicConfig(
level=LOG_LEVEL,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S%z",
)
log = logging.getLogger("gcs_sync")
# Duration parser: "90d", "24h", "30m", "45s"
_DURATION_RE = re.compile(r"^(\d+)([dhms])$")
_UNIT_SECONDS = {"d": 86400, "h": 3600, "m": 60, "s": 1}
def parse_duration(value: str) -> timedelta:
"""Parse '90d' / '24h' / '30m' / '45s' into a timedelta."""
match = _DURATION_RE.match(value.strip().lower())
if not match:
raise argparse.ArgumentTypeError(
f"Invalid duration '{value}'. Use e.g. 90d, 24h, 30m, 45s."
)
amount, unit = match.groups()
return timedelta(seconds=int(amount) * _UNIT_SECONDS[unit])
# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class FileMeta:
"""Metadata for a local or remote file used for sync decisions."""
path: str
size: int
mtime: datetime
@dataclass
class SyncPlan:
"""The set of operations to perform against the bucket."""
to_upload: list[tuple[Path, str]] = field(default_factory=list)
to_delete_mirror: list[str] = field(default_factory=list)
to_delete_age: list[str] = field(default_factory=list)
skipped: int = 0
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def is_excluded(rel_path: str, patterns: Iterable[str]) -> bool:
"""Return True if rel_path matches any glob pattern (fnmatch)."""
return any(fnmatch.fnmatch(rel_path, pat) for pat in patterns)
def walk_local(root: Path, excludes: list[str]) -> dict[str, FileMeta]:
"""Return relative-path -> FileMeta for every non-excluded local file."""
result: dict[str, FileMeta] = {}
root = root.resolve()
for dirpath, dirnames, filenames in os.walk(root):
# Prune excluded directories early for performance
rel_dir = os.path.relpath(dirpath, root)
dirnames[:] = [
d for d in dirnames
if not is_excluded(os.path.join(rel_dir, d).lstrip("./"), excludes)
]
for name in filenames:
full = Path(dirpath) / name
rel = str(full.relative_to(root)).replace(os.sep, "/")
if is_excluded(rel, excludes):
continue
try:
stat = full.stat()
except OSError as exc:
log.warning("Skipping unreadable file %s: %s", full, exc)
continue
result[rel] = FileMeta(
path=str(full),
size=stat.st_size,
mtime=datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc),
)
return result
def list_remote(bucket: storage.Bucket, prefix: str) -> dict[str, FileMeta]:
"""Return blob-name -> FileMeta for every object under prefix."""
result: dict[str, FileMeta] = {}
for blob in bucket.list_blobs(prefix=prefix or None):
name = blob.name
if prefix and name.startswith(prefix + "/"):
name = name[len(prefix) + 1:]
elif prefix and name == prefix:
continue
result[name] = FileMeta(
path=blob.name,
size=int(blob.size or 0),
mtime=blob.updated or datetime.now(timezone.utc),
)
return result
def build_plan(
local: dict[str, FileMeta],
remote: dict[str, FileMeta],
prefix: str,
mirror: bool,
delete_after: timedelta | None,
now: datetime,
) -> SyncPlan:
"""Compute what to upload and what to delete."""
plan = SyncPlan()
for rel, lmeta in local.items():
rmeta = remote.get(rel)
if rmeta is None:
plan.to_upload.append((Path(lmeta.path), rel))
elif rmeta.size != lmeta.size or rmeta.mtime < lmeta.mtime:
plan.to_upload.append((Path(lmeta.path), rel))
else:
plan.skipped += 1
if mirror:
local_keys = set(local.keys())
for rel in remote:
if rel not in local_keys:
plan.to_delete_mirror.append(rel)
if delete_after is not None:
cutoff = now - delete_after
for rel, rmeta in remote.items():
if rmeta.mtime < cutoff and rel not in plan.to_delete_mirror:
plan.to_delete_age.append(rel)
return plan
# ---------------------------------------------------------------------------
# Actions
# ---------------------------------------------------------------------------
def upload_one(
bucket: storage.Bucket,
local_path: Path,
blob_name: str,
chunk_size: int,
) -> tuple[str, bool, str]:
"""Upload a single file. Returns (blob_name, success, message)."""
try:
blob = bucket.blob(blob_name, chunk_size=chunk_size)
blob.upload_from_filename(str(local_path), timeout=600)
return blob_name, True, "uploaded"
except gcp_exceptions.GoogleAPICallError as exc:
return blob_name, False, f"API error: {exc}"
except OSError as exc:
return blob_name, False, f"local IO error: {exc}"
def delete_blob(bucket: storage.Bucket, blob_name: str) -> tuple[str, bool, str]:
"""Delete a single remote object."""
try:
bucket.blob(blob_name).delete(timeout=120)
return blob_name, True, "deleted"
except gcp_exceptions.NotFound:
return blob_name, True, "already gone"
except gcp_exceptions.GoogleAPICallError as exc:
return blob_name, False, f"API error: {exc}"
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
p = argparse.ArgumentParser(
description="Sync a local directory to GCS and enforce lifecycle rules.",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
p.add_argument("--bucket", required=True, help="Bucket name or gs:// URI")
p.add_argument("--local", required=True, type=Path, help="Local directory")
p.add_argument(
"--prefix", default=DEFAULT_PREFIX,
help="Remote key prefix (subfolder) inside the bucket",
)
p.add_argument(
"--delete-after", type=parse_duration, default=None,
help="Delete remote objects older than this (e.g. 90d)",
)
p.add_argument(
"--mirror", action="store_true",
help="Delete remote objects with no local counterpart",
)
p.add_argument(
"--exclude", action="append", default=[],
help="Glob pattern to exclude (repeatable)",
)
p.add_argument(
"--dry-run", action="store_true",
help="Print planned actions without executing them",
)
p.add_argument(
"--workers", type=int, default=DEFAULT_WORKERS,
help="Concurrent upload/delete workers",
)
return p.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
if not args.local.is_dir():
log.error("Local path %s is not a directory", args.local)
return 2
bucket_name = args.bucket.replace("gs://", "").rstrip("/")
prefix = args.prefix.strip("/")
try:
client = storage.Client()
except Exception as exc: # DefaultCredentialsError, etc.
log.error("Failed to create GCS client: %s", exc)
log.error("Set GOOGLE_APPLICATION_CREDENTIALS or run "
"`gcloud auth application-default login`.")
return 3
try:
bucket = client.bucket(bucket_name)
# Cheap existence check
bucket.reload()
except gcp_exceptions.NotFound:
log.error("Bucket %s does not exist or is not accessible", bucket_name)
return 4
except gcp_exceptions.Forbidden as exc:
log.error("Access denied to bucket %s: %s", bucket_name, exc)
return 5
log.info("Scanning local directory: %s", args.local)
local = walk_local(args.local, args.exclude)
log.info("Found %d local files (after excludes)", len(local))
log.info("Listing remote objects under prefix '%s'", prefix or "(root)")
try:
remote = list_remote(bucket, prefix)
except gcp_exceptions.GoogleAPICallError as exc:
log.error("Failed to list bucket: %s", exc)
return 6
log.info("Found %d remote objects", len(remote))
now = datetime.now(timezone.utc)
plan = build_plan(
local=local,
remote=remote,
prefix=prefix,
mirror=args.mirror,
delete_after=args.delete_after,
now=now,
)
log.info(
"Plan: upload=%d delete(mirror)=%d delete(age)=%d skipped=%d",
len(plan.to_upload), len(plan.to_delete_mirror),
len(plan.to_delete_age), plan.skipped,
)
if args.dry_run:
for path, name in plan.to_upload:
log.info("[DRY-RUN] UPLOAD %s -> %s", path, name)
for name in plan.to_delete_mirror:
log.info("[DRY-RUN] DELETE (mirror) %s", name)
for name in plan.to_delete_age:
log.info("[DRY-RUN] DELETE (age) %s", name)
return 0
# ---- Execute uploads ----
uploaded = failed_up = 0
if plan.to_upload:
with ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = [
pool.submit(
upload_one, bucket, path,
f"{prefix}/{name}" if prefix else name,
DEFAULT_CHUNK_SIZE,
)
for path, name in plan.to_upload
]
for fut in as_completed(futures):
name, ok, msg = fut.result()
if ok:
uploaded += 1
log.debug("Uploaded %s", name)
else:
failed_up += 1
log.error("Upload failed for %s: %s", name, msg)
# ---- Execute deletions ----
deleted = failed_del = 0
all_deletes = [
(n, "mirror") for n in plan.to_delete_mirror
] + [
(n, "age") for n in plan.to_delete_age
]
if all_deletes:
with ThreadPoolExecutor(max_workers=args.workers) as pool:
futures = {
pool.submit(
delete_blob, bucket,
f"{prefix}/{name}" if prefix else name,
): (name, reason)
for name, reason in all_deletes
}
for fut in as_completed(futures):
name, reason = futures[fut]
_, ok, msg = fut.result()
if ok:
deleted += 1
log.debug("Deleted (%s) %s", reason, name)
else:
failed_del += 1
log.error("Delete failed for %s: %s", name, msg)
log.info(
"Done. uploaded=%d deleted=%d skipped=%d failures=%d",
uploaded, deleted, plan.skipped, failed_up + failed_del,
)
return 1 if (failed_up or failed_del) else 0
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
log.warning("Interrupted by user")
sys.exit(130)
| Variable / Flag | Default | Purpose |
|---|---|---|
GOOGLE_APPLICATION_CREDENTIALS |
(unset) | Path to a service-account JSON key. If unset, Application Default Credentials are used (e.g. gcloud auth application-default login). |
GCS_SYNC_WORKERS / --workers |
8 |
Concurrent upload/delete threads. Raise for many small files; lower for large files or rate-limited buckets. |
GCS_SYNC_PREFIX / --prefix |
"" |
Remote subdirectory inside the bucket. Local paths are mapped relative to this prefix. |
GCS_SYNC_CHUNK_MB |
8 |
Resumable upload chunk size in MiB. Increase for very large files on fast links. |
GCS_SYNC_LOG_LEVEL |
INFO |
DEBUG shows every uploaded/deleted object; WARNING silences the summary. |
--delete-after |
(disabled) | Age threshold (90d, 24h, …). Only applies to remote objects older than the cutoff. |
--mirror |
off | Delete remote objects with no local counterpart. Destructive — combine with --dry-run first. |
--exclude |
(none) | fnmatch-style glob applied to relative paths. Repeatable. |
--dry-run |
off | Print the plan without touching the bucket. |
1. First-time upload with excludes and a dry-run preview:
$ python3 gcs_sync.py --bucket gs://my-data-bucket --local ./data \
--prefix prod/2024 --exclude '*.tmp' --exclude '.git/*' --dry-run
2024-06-01T10:00:01+0000 [INFO] Found 412 local files (after excludes)
2024-06-01T10:00:02+0000 [INFO] Found 0 remote objects
2024-06-01T10:00:02+0000 [INFO] Plan: upload=412 delete(mirror)=0 delete(age)=0 skipped=0
2024-06-01T10:00:02+0000 [INFO] [DRY-RUN] UPLOAD data/config.yaml -> prod/2024/config.yaml
...
2. Incremental sync + 90-day lifecycle cleanup:
$ python3 gcs_sync.py --bucket gs://my-data-bucket --local ./data \
--prefix prod/2024 --delete-after 90d --exclude '*.tmp'
2024-06-01T10:05:00+0000 [INFO] Found 412 local files (after excludes)
2024-06-01T10:05:01+0000 [INFO] Found 418 remote objects
2024-06-01T10:05:01+0000 [INFO] Plan: upload=3 delete(mirror)=0 delete(age)=17 skipped=409
2024-06-01T10:05:04+0000 [INFO] Done. uploaded=3 deleted=17 skipped=409 failures=0
3. Strict mirror (remote == local) with parallelism:
$ python3 gcs_sync.py --bucket gs://my-data-bucket --local ./data \
--prefix prod/2024 --mirror --workers 32
2024-06-01T10:10:00+0000 [INFO] Plan: upload=0 delete(mirror)=4 delete(age)=0 skipped=412
2024-06-01T10:10:02+0000 [INFO] Done. uploaded=0 deleted=4 skipped=412 failures=0
Exit codes: 0 success, 1 partial failures, 2 bad local path, 3 auth failure, 4 bucket not found, 5 forbidden, 6 listing error, 130 interrupted.
storage.Client() construction; prints actionable hint and exits 3.bucket.reload() raises NotFound or Forbidden; exits 4/5.walk_local, skipped rather than aborting the whole run.1). One bad file does not stop the batch.GoogleAPICallError and reported; the google-cloud-storage client already retries idempotent operations server-side.KeyboardInterrupt handler exits cleanly with 130; in-flight uploads may be partially written but GCS resumable uploads will overwrite on the next run.NotFound on delete is treated as success ("already gone").(size, mtime) to crc32c comparison — blob.crc32c vs google_crc32c.value_of_file(path) — for byte-exact guarantees across clock skew.delete_blob with blob.update_storage_class("NEARLINE") for age-based tiering, saving cost without losing data.SyncPlan to a JSON file so long-running syncs can resume after a crash without re-listing the bucket.--source-bucket flag and use bucket.copy_blob() to mirror between projects/regions.