skrawcz

import pytest

@pytest.fixture(scope="module")
def database_connection():
    """Fixture that creates a DB connection"""
    db_client = SomeDBClient()
    yield db_client
    print("\nStopped client:\n")

def test_my_function(database_connection):

skrawcz

import copy

from IPython.display import Image, display
from IPython.core.display import HTML 

from burr.core import ApplicationBuilder, State, default, graph, when
from burr.core.action import action
from burr.tracking import LocalTrackingClient

@action(reads=[], writes=["output_folder"])

skrawcz

import copy

from IPython.display import Image, display
from IPython.core.display import HTML 
import openai

from burr.core import ApplicationBuilder, State, default, graph, when
from burr.core.action import action
from burr.tracking import LocalTrackingClient

skrawcz

# python 3.12

"""
Hamilton demo. Runs the Hamilton code.
"""

import sys

import pprint

skrawcz

# %%cell_to_module config_plate --display
from hamilton.function_modifiers import parameterize, source, value

class TablePaths:
    pass # your code here

table_names = ["vendor", "predictions"] # update this


@parameterize(

skrawcz

# functions.py - declare and link your transformations as functions....
import pandas as pd
from hamilton.function_modifiers import parameterize_sources

def a(input: pd.Series) -> pd.Series:
    return input % 7

def b(a: pd.Series) -> pd.Series:
    return a * 2

skrawcz

@action(
    reads=["chat_history"],
    writes=["chat_history"],
)
def AI_generate_response(state: State, client: openai.Client) -> tuple[dict, State]:
    """AI step to generate the response."""
    messages = state["chat_history"]
    response = client.chat.completions.create(
        model="gpt-4-turbo-preview",
        messages=messages,

skrawcz

"""
This module demonstrates a telephone application
using Burr that:
 - captions an image
 - creates caption embeddings (for analysis)
 - creates a new image based on the created caption
"""
import os
import uuid

skrawcz

# functions.py - declare and link your transformations as functions....
import pandas as pd
from hamilton.htypes import Parallelizable, Collect


def motor(motor_list: list[int]) -> Parallelizable[int]:
   for _motor in motor_list:
      yield _motor

def _is_motor_on(motor: int ) -> bool:

skrawcz

from airflow.decorators import dag, task
from airflow.operators.python import get_current_context

# set the Airflow DAG parameters. This will appear in the Airflow UI. 
DEFAULT_DAG_PARAMS = dict(
    label="absenteeism_time_in_hours",
    feature_set=[
          "age_zero_mean_unit_variance",
          "has_children",
          "has_pet",