Skip to content

Instantly share code, notes, and snippets.

@leplatrem
Last active January 19, 2021 08:58
Show Gist options
  • Save leplatrem/f3cf7ac5b0b9b0b27ff6456f47f719ca to your computer and use it in GitHub Desktop.
Save leplatrem/f3cf7ac5b0b9b0b27ff6456f47f719ca to your computer and use it in GitHub Desktop.
import json
import os
from kinto_http import Client, BearerTokenAuth
SERVER = os.getenv("ENV", "prod")
SERVER_URL = {
"prod": "https://settings-writer.prod.mozaws.net/v1/",
"stage": "https://settings-writer.stage.mozaws.net/v1/",
"dev": "https://kinto.dev.mozaws.net/v1/",
}
AUTH = os.getenv("AUTH", "")
BUCKET = "main-workspace"
COLLECTION = "password-rules"
source_records = [...] # Fetch source of truth
client = Client(
server_url=SERVER_URL.get(SERVER, SERVER),
bucket=BUCKET,
collection=COLLECTION,
auth=tuple(AUTH.split(":", 1))
if ":" in AUTH
else BearerTokenAuth(AUTH),
)
records_by_id = {r["id"]: r for r in client.get_records()}
# Create or update the destination records.
to_create = []
to_update = []
for r in source_records:
record = records_by_id.pop(r["id"], None)
if record is None:
to_create.append(r)
elif not records_equal(r, record):
to_update.append(r)
# Delete the records missing from source.
to_delete = records_by_id.values()
has_pending_changes = (len(to_create) + len(to_update) + len(to_delete)) > 0
if not has_pending_changes:
print("Records are in sync. Nothing to do.")
return
with client.batch() as batch:
for r in to_create:
batch.create_record(data=r)
for r in to_update:
# Let the server assign a new timestamp.
del r["last_modified"]
batch.update_record(data=r)
for r in to_delete:
batch.delete_record(id=r["id"])
ops_count = len(batch.results())
# If collection has multi-signoff, request review, otherwise auto-approve changes.
server_info = client.server_info()
signer_config = server_info["capabilities"].get("signer", {})
signer_resources = signer_config.get("resources", [])
# Check collection config (sign-off required etc.)
signed_dest = [
r
for r in signer_resources
if r["source"]["bucket"] == BUCKET
and r["source"]["collection"] == COLLECTION
]
if len(signed_dest) == 0:
# Not explicitly configured. Check if configured at bucket level?
signed_dest = [
r
for r in signer_resources
if r["source"]["bucket"] == BUCKET
and r["source"]["collection"] is None
]
# Collection has no signoff features (eg. dev server). Nothing to do.
if len(signed_dest) == 0:
print(f"Done. {ops_count} changes applied.")
return
has_signoff_disabled = not signed_dest[0].get(
"to_review_enabled", signer_config["to_review_enabled"]
) and not signed_dest[0].get(
"group_check_enabled", signer_config["group_check_enabled"]
)
if has_signoff_disabled:
# Approve the changes.
client.patch_collection(data={"status": "to-sign"})
print(f"Done. {ops_count} changes applied and signed.")
else:
# Request review.
client.patch_collection(data={"status": "to-review"})
print(f"Done. Requested review for {ops_count} changes.")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment