rajathithan

#!/usr/bin/env python
# Python script to send email from cloud function using customized html template
# Author: Rajathithan Rajasekar 
# Version : 1.0
# Date: 07/30/2023

import os
import pytz
import smtplib
from datetime import datetime

rajathithan

[
  {
    "name":"XXXXX"
    "mode": "NULLABLE",
    "type": "RECORD",
    "fields": [
      {
        "name":"XXXXX"
        "mode": "NULLABLE",
        "type": "RECORD",

rajathithan

import datetime
from airflow import models
from airflow.operators.bash import BashOperator

YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

with models.DAG(
    dag_id="batch_test",
    schedule_interval=datetime.timedelta(days=1),
    start_date=YESTERDAY,

rajathithan

from __future__ import annotations
from typing import Any
import google.auth
from google.auth.transport.requests import AuthorizedSession
import requests
import functions_framework

AUTH_SCOPE = "https://www.googleapis.com/auth/cloud-platform"
CREDENTIALS, _ = google.auth.default(scopes=[AUTH_SCOPE])
web_server_url = "https://658a1a778d2442b9aac55afa02df7ab9-dot-us-east4.composer.googleusercontent.com"

rajathithan

from datetime import datetime
import pytz

import apache_beam as beam
from apache_beam.transforms import window

from log_elements import LogElements


with beam.Pipeline() as p:

rajathithan

import datetime
import pytz

import apache_beam as beam
from apache_beam.transforms import window

from log_elements import LogElements


class Event:

rajathithan

import apache_beam as beam

from log_elements import LogElements

with beam.Pipeline() as p:

  (p | beam.Create(['apple', 'banana', 'cherry', 'durian', 'guava', 'melon'])
     | beam.transforms.util.WithKeys(lambda x : x[0])
     | LogElements())
  

rajathithan

import apache_beam as beam

from log_elements import LogElements

with beam.Pipeline() as p:

  (p | beam.Create(range(400,500))
     | beam.combiners.Top.Largest(5)
     | LogElements())

rajathithan

import apache_beam as beam

from log_elements import LogElements

with beam.Pipeline() as p:

  (p | beam.Create(range(400,500))
     | beam.combiners.Top.Smallest(5)
     | LogElements())

rajathithan

import apache_beam as beam


class LogElements(beam.PTransform):

    class _LoggingFn(beam.DoFn):

        def __init__(self, prefix='', with_timestamp=False, with_window=False):
            super(LogElements._LoggingFn, self).__init__()
            self.prefix = prefix