xnuinside

def request_package_info_from_pypi(package_name):
    result = requests.get(
        f"https://pypi.org/pypi/{package_name}/json").json()
    info = result["info"]
    output = f'<i>{datetime.now().date()} Random Package from PyPi\n</i>' \
    f'<b>Package name:</b> {package_name}\n' \
    f'<b>Short description:</b> {info["summary"]}\n' \
    f'<b>Latest version:</b> {info["version"]}\n' \
    f'<b>Release date:</b> {result["releases"][info["version"]][0]["upload_time"].split("T")[0]}\n' \
    f'<b>Author:</b> {info["author"]}\n' \

xnuinside

while True:
   if date_ <= datetime.now().date():
       for i in range(5):
           if i == 4:
               date_ = datetime.now() + timedelta(days=1)

           bot.send_message(chat_id=chat_id,
                            text=request_package_info_from_pypi(
                                choice(bq_get_random_packages_downloaded_for_yesterday()))
                            )

xnuinside

@bot.message_handler(commands=['stats'])
def send_package_stats(message):
   package_name = message.text.split()[1].replace('_', '-')
   output = f"Stats for package {package_name} (numbers of downloads): \n"
   for i in range(4):
       date_ = (datetime.now().date() - timedelta(days=i+1)).isoformat()
       formatted_date_ = date_.replace('-', '')
       downloads = bq_get_downloads_stats_for_package(package_name, formatted_date_)
       output += f"<b>{date_}</b>: {downloads}\n"
   bot.reply_to(message, output, parse_mode="html")

xnuinside

import asyncio
from fastapi import BackgroundTasks

cur_folder = os.path.dirname(os.path.abspath(__file__))

def write_notification(email: str, message=""):
    with open(os.path.join(cur_folder, "log.txt"), mode="w+") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)

xnuinside

Introduction

Installing kubeflow on localmachine is not a simple task. Documentation on the official website might be outdated. At the time of writing, the solutions suggested include miniKF and microk8s. The later sets up GPU passthrough effortlessly.

• Install microk8s
• Enable microk8s features

xnuinside

🌱 If you interesting to participate in any project: feel free to open PR or 💬 issue, I very glad to any participation.

Right now, 🧑‍🌾 I'm focusing on supporting & adding new features to:

• simple-ddl-parser - Parser that focudes on differen DDL dialects and only DDL

• omymodels - Code generator that allows you to create different Python Modules from DDL and convert one model type to another

• py-models-parser - Parser that supported already 12 types of different Python models (ORM, Pydantic and etc)

xnuinside

def test_pydantic_models_generator_basic():

    ddl = """
CREATE table user_history (
        runid                 decimal(21) null
    ) ;
"""
    result = create_models(ddl, models_type="pydantic")["code"]

    expected = """from typing import Optional

xnuinside

@pytest.fixture
def load_generated_code():
    def _inner(code_text: str, module_name: Optional[str] = None) -> ModuleType:
        """method saves & returns new generated python module
        code_text - code to be saved in new module
        module_name: str - name of the module to use for saving the code
        """
            
        current_path = os.path.dirname(os.path.abspath(__file__))
        

xnuinside

def test_pydantic_models_are_working_as_expected(load_generated_code) -> None:
    ddl = """
    CREATE table user_history (
            runid                 decimal(21) null
        ,job_id                decimal(21)  null
        ,id                    varchar(100) not null -- group_id or role_id
        ,user              varchar(100) not null
        ,status                varchar(10) not null
        ,event_time            timestamp not null default now()
        ,comment           varchar(1000) not null default 'none'

xnuinside

import sqlparse


# put here in raw_sql your query
raw_sql = 'select column1, column2 from schema.table_name;'

# parse query
statements = sqlparse.parse(raw_sql)

# columns