Dminor7

import pendulum

def generate_intervals(start_datetime, end_datetime, interval_minutes=30):
    current_datetime = start_datetime
    while current_datetime <= end_datetime:
        yield current_datetime, current_datetime.add(minutes=interval_minutes)
        current_datetime = current_datetime.add(minutes=interval_minutes)

# Set your desired start and end datetime
start_datetime = pendulum.datetime(2023, 12, 12, 0, 0, 0, tz="UTC")

Dminor7

import pendulum

def generate_30min_time_range(timestamp_str):
    timestamp = pendulum.parse(timestamp_str)
    start_time = timestamp.start_of('hour')
    if timestamp.minute >= 30:
        start_time = start_time.add(minutes=30)
    result = []
    while start_time <= timestamp:
        result.append(start_time)

Dminor7

"""
Check this stackeroverflow question: https://stackoverflow.com/questions/68303327/unnecessary-list-item-nesting-in-bigquery-schemas-from-pyarrow-upload-dataframe
Check this github issue: https://github.com/googleapis/python-bigquery/issues/19
"""

from google.cloud.bigquery._pandas_helpers import *
from google.cloud.bigquery import _helpers
from google.cloud import storage
from google.cloud import bigquery

Dminor7

    SELECT 
    customer_id, 
    reduce(
	    event_vectors,
	    repeat(0.0, 512), -- dimension of vector here it is 512
	    (sum_array, element_array) -> zip_with(sum_array, element_array, (s, e) -> s + e), 
	    state_array -> transform(state_array, s -> s / cardinality(event_vectors))
    ) as mean_event_vector
    FROM(
        SELECT 

Dminor7

"""
Using a generator we can `yield` key and value pairs when the value is not an instance of `list` or `dict` 
else we recursively call the function to flatten the object. To preserve empty `list` and `dict` 
we add additional check on obj when checking the obj type.
"""


def flatten(obj, prefix=[], sep="_"):
    if isinstance(obj, list) and obj:
        for i, o in enumerate(obj):

Dminor7

from datetime import datetime
import uuid

from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook
from airflow.operators.python_operator import PythonOperator
from airflow.utils.state import State
from airflow import DAG

Dminor7

from datetime import datetime

from airflow import models
from kubernetes.client import models as k8s

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import (
    KubernetesPodOperator,
)

Dminor7

apiVersion: v1
kind: PersistentVolume
metadata:
  name: my-volume
spec:
  storageClassName: ""
  capacity:
    storage: 10G
  accessModes:
    - ReadOnlyMany

Dminor7

"""
# Worksheet header to data key mapping
header_to_key = {
    "Name":"name",
    "Age":"age",
    "Weight(Kg)":"weight_kg",
}
# List of objects to load into worksheet
data = [
    {

Dminor7

import gspread_dataframe as gd
import gspread as gs
gc = gs.service_account(filename="credentials.json")

def export_to_sheets(sheet_name,df,mode='r'):
    ws = gc.open("Hashnode").worksheet(sheet_name)
    if(mode=='w'):
        ws.clear()
        gd.set_with_dataframe(worksheet=ws,dataframe=df,include_index=False,include_column_header=True,resize=True)
        return True