IL-ADAM

from ruamel.yaml import YAML
import json
from ruamel.yaml.scalarstring import PreservedScalarString
from typing import Text, Optional, Dict, Any, List


def read_json_data_contract(data_contract_path: Text) -> Dict[str, Any]:
    """
    Read JSON data contract from a file.

IL-ADAM

import typer
import os
import glob
from cli.apps.libs.dbt_contracts.generate_dbt_model import generate_model_yaml
from typing_extensions import Annotated
from typing import Tuple, List, Text
from pathlib import Path

CONTRACT_DIR = f"yourpath/data_contracts/users"

IL-ADAM

version: 2
models:
- name: exmaplemodel
  description: |-
    Data contract for user records generated by the Example App

    [DataContract on our GitHub repo](https://github.com/YourOrg/YourProject/yourpath/data_contracts/users/user_v1.json)
  config:
    tags:
    - infinitelambda

IL-ADAM

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "$id": "user_v1.json",
  "type": "object",
  "title": "User data contract",
  "description": "Data contract for user records generated by the Example App",
  "tags": [
    "infinitelambda",
    "user",
    "exmapleapp"

IL-ADAM

name: ksqlDB Migrations
on:
  workflow_call:
    inputs:
      environment:
        type: string
        description: Environment to deploy to
        required: true

jobs:

IL-ADAM

CREATE STREAM IF NOT EXISTS user_merged_stream WITH (
    KAFKA_TOPIC = '{{ cookiecutter.user_merged_topic }}',
    VALUE_FORMAT = 'JSON'
  )
  AS
  SELECT
    REPLACE(`user_id`, '"', '') as KEY,
    AS_VALUE(REPLACE(`user_id`, '"', '')) as `user_id`,
    `first_name` STRING,
    `last_name` STRING,

IL-ADAM

CREATE STREAM IF NOT EXISTS user_outbound_stream(
    `user_id` STRING KEY,
    `first_name` STRING,
    `last_name` STRING,
    `app_id` STRING,
) WITH (
    KAFKA_TOPIC = '{{ cookiecutter.user_outbound_topic }}',
    VALUE_FORMAT = 'JSON_SR'
  );
SET 'auto.offset.reset' = 'earliest';

IL-ADAM

CREATE STREAM IF NOT EXISTS user_inbound_stream(
    `user_id` STRING KEY,
    `first_name` STRING,
    `last_name` STRING,
    `app_id` STRING,
) WITH (
    KAFKA_TOPIC = '{{ cookiecutter.user_inbound_topic }}',
    VALUE_FORMAT = 'JSON_SR'
  );
SET 'auto.offset.reset' = 'earliest';

IL-ADAM

default_context:
    project_slug: ./migrations
    environment: CONFLUENT_ENVIRONMENT  # Value replaced during CI/CD

IL-ADAM

{
  "project_slug": "./migrations",
  "environment": "dev",
  "region" : "eu-west-1",
  "app" : "infinitelambda",
  "user_inbound_topic": "{{ cookiecutter.environment }}-{{ cookiecutter.app }}-user.inbound-{{ cookiecutter.region }}",
  "user_outbound_topic": "{{ cookiecutter.environment }}-{{ cookiecutter.app }}-user.outbound-{{ cookiecutter.region }}",
  "user_merged_topic": "{{ cookiecutter.environment }}-{{ cookiecutter.app }}-user.merged-{{ cookiecutter.region }}",
}