Kilo59

"""
pip install fastapi uvicorn

uvicorn dep_dec_api:app --reload
"""
import base64
import enum
import logging
from collections.abc import Mapping
from pprint import pformat as pf

Kilo59

import great_expectations as gx

context = gx.get_context()

my_datasource = context.get_datasource("my_datasource")
print(f"{my_datasource!r}")

# update the connection string (or any other property)
my_datasource.connection_string = "my_protocol://user:password@host/db"

Kilo59

from __future__ import annotations

import pathlib
import sys
from collections import Counter
from pprint import pformat as pf
from typing import Iterable, Iterator, NamedTuple

from mypy import api
from typing_extensions import override

Kilo59

from great_expectations.datasource.fluent import SQLDatasource
import sqlalchemy.engine
from typing import Literal, Optional


class CustomSQLDatasource(SQLDatasource):
    type: Literal["my_custom_sql"] = "my_custom_sql"
    connection_string: Optional[str] = None  # this makes this field optional
    some_new_param: dict = {}

Kilo59

Kilo59

Kilo59

[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[packages]
great-expectations = "==0.15.36"
pydantic = "==1.10.2"
typing_extensions = "==4.4.0"
# used to get package versions

Kilo59

# ##############
# at what number does this error?
# 0? 100,000? never?
# ##############
try:
    for i, n in enumerate(range(0, 100000)):
        assert i is n, f"{i} is {n}"
except AssertionError as e:
    print(f"Error on -> {e}")

Kilo59

import asyncio
import time

import httpx


async def print_joke(r: httpx.Response):
    r.raise_for_status()
    await r.aread()
    value = r.json()["value"]

Kilo59

# assuming pipx is already installed
# https://pipxproject.github.io/pipx/installation/

# airflow needs a home, ~/airflow is the default,
# but you can lay foundation somewhere else if you prefer
# (optional)
export AIRFLOW_HOME=~/airflow

AIRFLOW_VERSION=2.0.1
PYTHON_VERSION="$(python3 --version | cut -d " " -f 2 | cut -d "." -f 1-2)"