johschmidt42

# Databricks 13.3 LTS
# 
# - Python 3.10.12 (minor version must match with your Python version)
# - databricks-connect >=13.3 (uninstall pyspark first!)


from databricks.connect import DatabricksSession
from pyspark.sql import DataFrame
from pydantic_settings import BaseSettings

johschmidt42

import json
from typing import List

from azure.functions import AuthLevel, FunctionApp, HttpRequest, HttpResponse

app: FunctionApp = FunctionApp()


@app.function_name(name="GetRoles")
@app.route(route="GetRoles", methods=["POST"], auth_level=AuthLevel.ANONYMOUS)

johschmidt42

---
name: "Lint PR"
on:
  pull_request_target:
    types:
      - opened
      - edited
      - synchronize
jobs:
  main:

johschmidt42

import uuid

from databricks.databricks_service import DatabricksService, SecretsConfig


def create_job_run_on_new_cluster():
    secret_config: SecretsConfig = SecretsConfig(_env_file="../.env")
    databricks_service: DatabricksService = DatabricksService(secret_config)
    job_id: str = str(uuid.uuid4())
    image_url: str = "databricksjobrunacr.azurecr.io/databricks-wheel-scripts:latest"

johschmidt42

def create_job_run_on_new_cluster(
    self,
    job_id: str,
    image_url: str,
    package_name: str,
    entry_point: str,
    positional_arguments: Optional[List[str]] = None,
    named_arguments: Optional[Dict[str, str]] = None,
    notification_ids: Optional[List[str]] = None,
    env_vars: Optional[dict] = None,

johschmidt42

import logging
from json import JSONDecodeError
from typing import Dict, List, Optional

import httpx
from httpx import Response
from pydantic import BaseSettings

logger = logging.getLogger(__name__)

johschmidt42

FROM databricksruntime/python:12.2-LTS as base

WORKDIR /app

# ignore 'Running pip as the root user...' warning
ENV PIP_ROOT_USER_ACTION=ignore

ENV PATH="/databricks/python3/bin:${PATH}"

# update pip

johschmidt42

#!/Users/johannes/learnspace/databricks-python-wheel-job-run/.venv/bin/python
# -*- coding: utf-8 -*-
import re
import sys
from dbscript2.__main__ import main
if __name__ == '__main__':
    sys.argv[0] = re.sub(r'(-script\.pyw|\.exe)?$', '', sys.argv[0])
    sys.exit(main())

johschmidt42

from typer import Typer

app: Typer = Typer()


@app.command()
def script2(argument: str) -> None:
    print(__file__)
    print(f"Your argument is: {argument}")

johschmidt42

from typer import Typer

app: Typer = Typer()


@app.command()
def script1(argument: str) -> None:
    print(__file__)
    print(f"Your argument is: {argument}")