Ну что сегодня пятница а значит время продолжить большой разбор #Airflow 🚀

Сегодня в большом видео примере вы узнаете:

1. Как писать свой первый #DAG.
2. В чем разница между requests.get() и requests.Session().get() она есть и приличная.
3. Напишем на #python красивый класс для API #CoinMarketCap.
4. Запустим наш даг, получим ошибку и еще разок запустим 😎


Также напоминаю что весь код лежит тут в репозитории, он публичный и доступен для Вас.

Пересылайте друзьям и коллегам кто хочет научится писать свой первый даг но не знает с чего начать 📣

Любые вопросы в комментариях, всегда к вашим услугам.

В следующем видео сделаем #CICD и установим #Postgres на наш сервер чтобы там хранить данные не пропускайте! 🙄

#DEmeme

Сколько дашбордов не делай, а бизнесс все равно в эксель смотрит 😶

#DEMeme
Слово AI из каждого утюга, возможно даже сам утюг под собой скрывает нейросеть ⚙️

В этом меме много боли на самом деле 😄

#PyTrickMonday

С началом новой недели дорогие подписчики!
Сегодня расскажу как сделать собственный контекст-менеджер через contextlib.contextmanager

Уверен вы прекрасно знакомы с конструкцией with open(file, 'r') as file_to_open:
Или при работе с базой with engine.connect() as conn:
А вот как сделать свой собственный контекстный менеджер? Сейчас все покажу! 💡

Для начала что такое менеджер контекста with?
Это такой специальный объект имеющий два магических метода - магия в том что они прячутся за словом with и вы их не видите 👀

__enter__()   # запускается при входе в with-блок  
__exit__()    # вызывается при выходе (даже если внутри случился Exception)

Для примера давайте перепридумаем функцию open()

from contextlib import contextmanager # импорт контекстного менеджера

@contextmanager
def open_file(path, mode="r"):
    print("открывашка")
    f = open(path, mode)
    try:
        yield f # всё, что до yield это __enter__; все что после это 

ачалом нов

    finally:
        print("закрывашка")
        f.close()


with open_file("send_log.txt") as f: # наш собственный контекстный оператор
    data = f.readline()
    print(data)

# Вывод:
открывашка
2025-05-30 16:24:14,565 | INFO | Connecting to 149.154.167.91:443/TcpFull...
закрывашка

Самое важное что тут надо понять что там где yield - это то что вы делаете в теле контекста 😃

Давайте что то более жизненное разберем.
Например "Я хочу знать сколько памяти мой датафрейм съел после открытия файла большого"
Вот как это можно красиво сделать:

import os, psutil
from contextlib import contextmanager

@contextmanager
def mem_used(name: str = "block"):
    proc   = psutil.Process(os.getpid()) # берет текущий процесс
    start  = proc.memory_info().rss # размер в bytes
    try:
        yield # вот тут вы откроете ваш файл
    finally:
        end   = proc.memory_info().rss # было памяти у процесса в начале
        diff  = end - start  # сколько съело ваше чтение bytes
        mb    = 1024 ** 2                     # bytes в MiB

        print(f"MemoryUsed: {diff/mb:+.2f} MiB")


with mem_used("parse_json"):
    big_df = pd.read_json("og.json", lines=True, orient="records")

# MemoryUsed: +1.17 MiB

Пересылайте коллегам, чтобы они наконец-то они узнали что же за магия скрывается за WITH 🪄

Теперь домашнее задание напишите контекстный менеджер считающий время выполнение вашего кода with timer("любое имя процесса")
вывод должен быть такой "Имя процесса: время в секундах"
Присылайте решения в комменты 🔽

Делаем сами без помощи GPT! в этом посте все есть, чтобы вы справились, разогрейте мозги в начале недели 🧠

2️⃣🔤🔤🔤🔤🔤 без ИИ #DEInsight

ИИ технологии настолько сильно проникли в мою жизнь что уже и не припомню когда бы был хоть день, когда я не задавал какой то вопрос GPT.
И мне стало интересно "А что если я на работе вообще не буду пользоваться ИИ"
Вот что я понял за 2 недели, работая без ИИ помошников(GPT, co-Pilot). Будет душновато отркываем форточки 😄

0️⃣Все ответы есть в ДОКУМЕНТАЦИИ или в Reddit или в StackOver. Читайте доку и вам не придется пытать и оскорблять ГПТ пока он не даст ответ 🙂
1️⃣Получая ответ от ГПТ я просто использовал, не запоминая особо, а через неделю мог спросить тоже самое легко, но прочитав доку, поресерчив в интренете и найдя ответ он остался в моей памяти и за две недели я освежил много уже забытых знаний и получил кое что новое. Я это "выстрадал" поэтому оно отпечаталось.
2️⃣Использование ИИ создает ложное ощущение что твоя экспертиза растет ведь ты можешь сделать гораздо больше. Но это иллюзия и самообман, когда ты не провел ресерч должным образом, а просто выбрал "думающую" модель для лучшего ответа она сама за тебя почитала доку и дала тебе решение, твой вклад в это только задданый вопрос.
3️⃣Чтобы запоминать и получать опыт надо иметь взаимосвязь: Выявление проблемы - Поиск ответа(чтение доки, редита) - Использование найденных ответов -- после нескольких итераций таких -- Находишь решение и объяснение этому решению. И весь этот процесс ресерча создает прочные нейронные связи и это тот же дофамин но более сложный. А ИИ за вас делает ресерч и выдает решение и когда у "вас" получилось вы тоже получаете дофамин, но быстрый - и это проблема.
3️⃣🔤1️⃣ Проведите аналогию между съесть шоколадку и полежать дома сейчас чтобы стало хорошо, или продолжать следить за питанием и заниматься спортом чтобы отлично выглядеть но потом.
Вы получите удовольствие в обоих случаях, но во втором это более сложный путь и он скучнее, но для вашего здоровья и фигуры гораздо более правильный выбор да и опыт который вы получите куда более применим чем шоколад и диван 🙂
Использование ИИ в решении ваших рабочих(любых) задач это таже ловушка и с этим надо быть аккуратнее.
4️⃣Есть иллюзия что ИИ помогает быстрее решать задачи, это не правда. Скорость только в начале упала, но "поскрипя" пару дней быстро восстанавливаешься. Потому что то что раньше делалось на автомате, потом делегировалось ИИ и поэтому пришлось "поскрипеть".
5️⃣У меня вновь появилось желание изучать новое, оно как будто бы заснуло на время.
6️⃣В первые два дня я не мог написать рекурсивную функцию примерно час 😄 А раньше такое делалалось изи. Я мог и не писать ее, но мне хотелось.
7️⃣Критическое мышление с ИИ притупляется - факт. Обычно у вас так - "ой че то он херь пишет, сам сделаю". А это уже звоночек что, не получая рузультата, такого как надо вам вы уже не тратите время на осмысления, почему он пошел не туда. Или может вы его туда завели 😄просто потому что не углубились в проблему которую решаете.
8️⃣Английский. Я так ленился сам читать и переводить и писать, что давал ГПТ и говорил ответь вот что и просто копировал отсылая обратно. Поделав это самостоятельно вновь, я понял что опять же у меня была иллюзия что это "Я" общаюсь, потому что я писал в промпт что надо перевести на английский. Но это вообще не я и это не мои слова я просто в очередной раз просил ГПТ решить задачу за меня. Даже не техническую.


Вывод какой напрашивается:
ИИ это прекрасно, но надо делать передышку и вернуть ресерч в вашу жизнь. Иначе вы буквально тупеете, а думаете что эксперт 🤷‍♀️
После пары недель я чувствую прилив знаний и дофамина от решения задач самостоятельно.
Быстро понимаешь где у тебя реальные знания есть(или были) а где ГПТ-завеса.
Когда думал что экономил время на самом деле экономил мозги.


Такие философские рассуждения.
Это лишь мое мнение - мой опыт(НЕ ИСТИНА), вы можете думать иначе, пишите в коментариях что думаете 🪄

#SQLWednesday

Сегодня обсудим классический вопрос с собеседования для #DataEngineer. 🪄
Вопрос звучит так:
"Перед тобой #csv файл 20ГБ, как откроешь?"
Конечно никакой pd.read_csv() тут не поможет это просто уложит ваше ядро в #JupyterNotebook.
-- Или поможет ? Дочитай до конца и узнаешь.
Такой вопрос могут задать на самом деле и Junior и Middle да и выше. Потому что не всегда очевидно как это сделать красиво и эффективно.

Самый эффективный способ работать с такого размера файлом это естесвенно загрузить его в базу.
#Postgres уже имеет нативный инструмент для работы с csv файлами и вот как это выглядит.

COPY raw.sales
FROM '/var/lib/postgresql/big_sales.csv'   -- путь до файла на сервере или локальный путь(если постгре локальная).
WITH (
    FORMAT      csv,       -- CSV-парсер
    HEADER      true,      -- пропускаем строку заголовков
    DELIMITER   ',',       -- если вдруг не запятая
    QUOTE       '"',       -- кавычки по умолчанию
    ENCODING    'UTF8'     -- кодировка можно изменить если будет "криво"
);

Легко и просто. Но мы не всегда имеем доступ к серверной файловой системе, так вот как это сделать прям из JupyterNotebook не ломая ядро?

1. Для нашего эксперимента я сгенерировал файл гигант вес которого 13.4GB
2. У меня уже есть развернутый постгрес на удаленном сервере(вы можете локально установить)
3. Будем пользоваться любимым пандас с chunksize

import pandas as pd
import sqlalchemy as sa

DB = dict(
    user="",
    password="",
    host="",
    port=5432,
    dbname="",
)
engine = sa.create_engine(
    f"postgresql+psycopg2://{DB['user']}:{DB['password']}@{DB['host']}:{DB['port']}/{DB['dbname']}",
)

#  параметры CSV
csv_path   = "big_file.csv"
chunk_rows = 100_000 # будем грузить по 100K строк
table_name = "example_csv"
schema     = "raw"

# создаём таблицу один раз с нужными колонками
with engine.begin() as conn:
    conn.exec_driver_sql(f"""
        CREATE SCHEMA IF NOT EXISTS {schema};
        DROP TABLE IF EXISTS {schema}.{table_name};
        CREATE TABLE {schema}.{table_name} (
            col_0 text, col_1 text, col_2 text, col_3 text, col_4 text,
            col_5 text, col_6 text, col_7 text, col_8 text, col_9 text
        );
    """)

# грузим чанками
reader = pd.read_csv(
    csv_path,
    chunksize=chunk_rows,
    iterator=True,
    header=0,
)

for i, chunk in enumerate(reader, 1):
    chunk.to_sql(
        name=table_name,
        con=engine,
        schema=schema,
        if_exists="append",
        index=False,
        method="multi",
    )
    print(f"Чанк {i}: {len(chunk)} строк залито")

print("Готово!")

Пересылайте коллегам, чтобы не терялись на собеседованиях и знали что ответить 💡

Чтобы поиграть дома в DE тетрадку с кодом прилагаю 🧠

#DataJungle #Postgres #ETL

Сегодня раворачиваем #Postgres, настраиваем #CICD, выходим из #vim !🤝

Это финальное видео в серии о том как деплоить #Airflow и делать свои даги.
Как и обещал рассказал про CICD и о том как развернуть postgres, чтобы наши данные было бы где хранить.

Как разворачивать postgres в #ubuntu как настоящий девопс?

1. Устанавливаем саму базу:

sudo apt install postgresql

2. Разрешаем слушать внешние адреса файл ( /etc/postgresql/*/main/postgresql.conf ):

 
listen_addresses = '*'

3. Разрешаем коннект именно с вашего IP(файл /etc/postgresql/*/main/pg_hba.conf ) на нижней строке добавляем.

host DB_NAME       USSERNAME        YOUR_IP        md5

4. Заходим в CLI psql создаем базу и пользователя:

sudo -u postgres psql

CREATE DATABASE DB_NAME;
CREATE USER USERNAME WITH PASSWORD 'PASSWORD';

Все на этом базовый деплой базы будет завершен. Все остальные подробности в видео!

Атракцион невиданной щедрости! 🔔
И самое главное, я обещал что вы поиграете в реального #DataEngineer что для этого надо:
1. Быть подписанным на классный канал про дата инжиниринг @data_jungle.
2. Быть в чате этого канала.
3. В комментарии к этому посту написать "хочу поиграть в DE ВАШ НИК В ГИТХАБ и ВАШ IP ноута с которого работаете".

После этого вы получите
- инвайт в репозиторий
- крэды в Airflow
- крэды в БД на ЧТЕНИЕ(это значит что инсерт вы сможете сделать только через ДАГ, а читать можете с jupyter из дома).

Это позволит вам со своей ветки пушить код в наш скромный продакшн 👀 я буду лично проводить ревью кода и писать комментарии чтобы вы могли становится лучше.

Правила игры:
- нельзя использовать метод pandas.to_sql все делаем только через хук как на видео, предварительно создавайте таблицы через таску в #airflow в схеме raw_data.
- данные можете качать какие угодно, любой #API который нравится, просто добавляйте ваш ключ в переменные и пользуйтесь.
- нельзя хранить больше 1М строк в вашей таблице в базе.

Не бойтесь уронить Airflow, вы учитесь это нормально, задавайте вопросы не стейсняйтесь, и да прибудет с вами сила 🔫

#PyTrickMonday

С началом новой недели друзья, сегодня 3 нереально крутых приёма в #pandas, про которые большинство даже не догадываются 👀

1️⃣Series.explode() - превращаем «списки в ячейках» в длинную таблицу одной строчкой. Очень полезная штука если например у вас есть вот такой объект когда к 1 заказу есть два айтема и вы бы хотели получить такой вид:
order_id items
0 101 BTC
1 101 ETH
2 102 DOGE

import pandas as pd

orders = pd.DataFrame({
    "order_id": [101, 102],
    "items": [["BTC", "ETH"], ["DOGE"]]
})
flat = orders.explode("items", ignore_index=True)

Просто используйте метод explode() для вашей колонки с вложенными данными ✔️

2️⃣ #DataFrame .pipe() - конвейер #DataEngineering по взрослому.

Pandas любит цепочки вызовов вместо лапши кода. Если вам надо применить к датафрейму много функций то можно выстроить это в pipeline очень легко и просто.

def add_profit(df):
    return df.assign(profit=df.sales - df.cost)

def top(df, n=5):
    return df.nlargest(n, "profit")

df = (pd.read_csv("shop.csv")
        .pipe(add_profit) # добавляем новую колонку профит
        .pipe(top, n=3))  # берем топ 3 по профиту

# передаёшь прямо в pipe свои аргументы функций

Очень удобно, меньше кода больше эффективности ✔️

3️⃣DataFrame.convert_dtypes() - минус RAM в одну команду.
Загружаешь CSV - и видишь большинство типов object и для дальнейшей работы надо делать преобразование типов .astype().

raw   = pd.read_csv("huge.csv")
clean = raw.convert_dtypes()

# переводит «строки» в современный тип string[pyarrow] или string
# меняет int64 на nullable Int64, NaN больше не проблема
# падает объём памяти до ×4 раз (часто - и больше)

Одна строчка экономит кучу памяти ✔️

Напиши в комментариях сколько удалось сэкономить места с .convert_dtypes() проверь на каком то жирном DataFrame🧐

Пересылай друзьям и коллегам, уверен они будут благодарны тебе 👀

#SQLWednesday

Сегодня будем разбирать задачу с собеседования #DataEngineer или #DataAnalyst задача покажет хорошее знание SQL а также ваши способности как аналитика!
Итак ДАНО:

😀У нас есть таблица поездок на самокатах.
Нужно посчитать, сколько в среднем простаивает самокат между поездками каждый день.
Вот так выгледят данные:

scooter_id | ride_id | user_id |   ride_start_time
       101 | R1      |    2001 | 2024-06-01 09:00:00
       101 | R2      |    2002 | 2024-06-01 11:00:00
       101 | R3      |    2003 | 2024-06-01 15:00:00
       101 | R4      |    2004 | 2024-06-02 10:30:00

Задачу можно сформулировать вот так: "для каждого самоката и каждой даты посчитать среднее время простоя между поездками (в минутах или часах — на твой вкус 😎)"
Давайте решать
(я не сторонник CTE) поэтому сначала давайте напишем запрос который бы показал предыдущее время поездки каждого скутера и поможет нам в этом функция #LAG:

  SELECT
    scooter_id,
    ride_start_time,
    DATE(ride_start_time) AS ride_date,
    LAG(ride_start_time) OVER (PARTITION BY scooter_id ORDER BY ride_start_time) AS prev_start_time
  FROM scooter_rides

Что получим(не очень красиво отображается, убрал в выводе ride_date):

| scooter_id | ride_start_time     | prev_start_time     |
|------------|---------------------|---------------------|
| 101        | 2024-06-01 09:00:00 | NULL                |
| 101        | 2024-06-01 11:00:00 | 2024-06-01 09:00:00 |
| 101        | 2024-06-01 15:00:00 | 2024-06-01 11:00:00 |
| 101        | 2024-06-02 10:30:00 | 2024-06-01 15:00:00 |

Это именно то что мы хотели, чтобы посчитать разницу между поездками нам нужна колонка предыдущей поездки ⚡️
Что же делает LAG - смещает вашу колонку на единицу на основании PARTITION BY. Кстати в LAG(column, offset) вторым параметром можно передать на сколько строк вы хотите сместится по умолчанию это 1.
Уверен вы уже поняли как решать, но позвольте закончить - я сразу посчитаю разницу между текущем и предыдущим временем и сразу извлеку #EPOCH (секунды чтобы легко конвертировать в минуты):

SELECT
    scooter_id,
    ride_date,
   ROUND(AVG(downtime_minutes), 1) AS avg_downtime_minutes
FROM (
    SELECT
        scooter_id,
        DATE(ride_start_time) AS ride_date,
        EXTRACT( EPOCH FROM ride_start_time  - LAG(ride_start_time) OVER (PARTITION BY scooter_id ORDER BY ride_start_time)) / 60 AS downtime_minutes
    FROM scooter_rides
) AS t
WHERE downtime_minutes IS NOT NULL -- чтобы пропустить строку с нулом
GROUP BY scooter_id, ride_date
ORDER BY scooter_id, ride_date;

Вуаля

| scooter_id | ride_date   | avg_downtime_minutes |
|------------|-------------|----------------------|
| 101        | 2024-06-01  | 180.0                |
| 101        | 2024-06-02  | 1170.0               |

Важно что надо помнить при решении таких задач ВСЕГДА начинайте с внутреннего запроса с оконной функцией, потому что сформировав его вы уже решили задачу на 70%.

Пересылайте друзьям и коллегам, сохраняйте себе в избранное в телеграмме чтобы не забыть ☺️
Можете ли вы решить задачу как то еще пишите в комментариях 😄
#DataJungle #SQL

#PyTrickMonday

С началом новой недели друзья! Самое время разобрать парочку полезных #python трюков 🐍

1️⃣Универсальная распаковка (* и **)
Как по-красоте «разворачивать» коллекции и аргументы? Уверен вы уже видели эти загадочные звезды в коде

def connect(host, port, user, password):
    print(f"Connecting to {host}:{port} as {user}")

cfg = {"host":"localhost","port":5432}
auth = {"user":"postgres","password":"secret"}

# Объединяем словари и распаковываем сразу в функцию
connect(**{**cfg, **auth})
# Connecting to localhost:5432 as postgres

# Распаковка списка в позиционные аргументы
coords = [10.0, 20.0, 30.0]
print(*coords)  # 10.0 20.0 30.0

# Для настоящего сигмы 
first, *middle, last = ["start", "step1", "step2", "end"]
print(first, middle, last)  # start ['step1','step2'] end

Правила такие:
- * собирает или пакует(распакует) в лист кусочек вашего итерируемого объекта даже строки:

>>> s='1234'
>>>first, *mid, last = s
>>> first
'1'
>>> mid
['2', '3', '4']
>>> last
'5'

- ** именованные параметры работает только для словарей, очень удобно передавать в функции или классы как объект.

def log_event(event_type, **details):
    print(f"Event: {event_type}")
    for key, value in details.items():
        print(f"  {key}: {value}")

# Вызов с произвольными именованными аргументами
log_event(
    "user_login",
    user_id=123,
    timestamp="2025-06-15T10:00:00",
    ip="192.168.0.1"
)

# Вывод:
# Event: user_login
#   user_id: 123
#   timestamp: 2025-06-15T10:00:00
#   ip: 192.168.0.1

2️⃣Типизированные датаклассы #dataclass
Идеально для работы с API или объектами из базы данных. Идеальное решения для бэкенда.

from dataclasses import dataclass

@dataclass
class User:
    id:    int
    name:  str
    email: str

# Плюс: с Python 3.10+ можно явно указать List[str], Optional[int] и т.д.

u = User(id=1, name="Alice", email="[email protected]")
print(u)
# User(id=1, name='Alice', email='[email protected]')

И тут напршивается а что если использовать распаковку ** с дата классами? Да легко!

# Предположим, получаем параметры из API вот в таком виде и нам надо дальше с ними удобно работать.
data = {"id":2,"name":"Bob","email":"[email protected]"}
user = User(**data)  # Всё в одну строку!

# получаем доступ к любому элементу 
user.id, user.name # 2, 'Bob'

Что мы получаем зная такие классные трюки?

- */** вы избавляетесь от ручного «распихивания» параметров.
- Единая точка правды для данных, dataclass позволяет задать структуру данных в одном месте, а затем
распаковывать в неё имеющиеся словари и доставать данные.
✅ Никаких «лишних» проверок типов и ручного присвоения - всё берётся из аннотаций.
✅ Ваша модель данных всегда синхронизирована: если вы добавите новое поле, IDE и линтер сразу подскажут места, где нужно передать или обработать его.

Где используем?

Конфигурации: храните настройки в YAML/JSON, загружайте как словарь и сразу распаковывайте в @dataclass.
API-клиенты: передавайте параметры запросов или заголовков одним **kwargs, избегая многословных вызовов.
Модели данных: вместо голых словарей для бизнес-логики оперируйте @dataclass - IDE подскажет нужные атрибуты и защитит от опечаток.

Пересылайте коллегам чтобы они уже сегодня использовали 👀

Привет дорогой подписчик!
Я был в отпуске 🏝️
с завтра посты будут выходить вновь регулярно
По понедельникам змеиные трюки #PyTrickMonday
По средам разбираем SQL #SQLWednesday
И конечно большие разборы интересных тем 😄

#PyTrickMonday
С началом новой недели, друзья! 🔥
Сегодня разбираем одну из самых «незаметных», но реально полезных фич - паттерны с #enumerate и #zip.
Уверен вам всем знаком трюк с enumerate:

items = ["BTC", "ETH", "SOL"]
for idx, symbol in enumerate(items, start=1):
    print(f"{idx}. {symbol}")

- start=1 - чтобы начать нумерацию с единицы (по умолчанию с 0).
- Избавляет от классического костыля с range(len(...)).

❓А можно ли параллельно итерироваться по двум спискам ? zip()

pairs = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
prices = [100000, 3000, 56] #цена выдумана любое совпадение случайно

for pair, price in zip(pairs, prices):
    print(f"{pair}: {price}$")

- Если один список короче - лишние элементы просто отбросятся.
- Чтобы «паддинг» сделать, можно использовать itertools.zip_longest.

А если надо по трем спискам учитывать индекс и еще и за один проход цикла?

dates = ["2025-06-30", "2025-07-01", "2025-07-02"]
volumes = [1234, 2345, 3456]
pairs = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
prices = [100000, 3000, 56] #цена выдумана любое совпадение случайно

for i, (pair, price, vol) in enumerate(zip(pairs, prices, volumes), start=1):
    print(f"{i}) {pair} — {price}$, объём {vol}")

- Оборачиваем zip в enumerate, чтобы сразу иметь и счётчик, и все значения.
- Кортеж (pair, price, vol) можно сразу «распаковать».

Где же использовать такую комбинацию zip + enumerate?

1. Самое частое где я пользовался этим в разработке это конечно тесты чтобы сопоставить тест кейс и результат.

inputs  = [2, 4, 6]
outputs = [4, 16, 36]

for test_num, (inp, outp) in enumerate(zip(inputs, outputs), 1):
    assert my_func(inp) == outp, f"Fail on test #{test_num}"

2. Логика с порогами для разных метрик идеально вообще

metrics = ["CPU", "RAM", "Disk"]
values  = [75, 60, 90]
limits  = [80, 70, 85]

for i, (m, v, lim) in enumerate(zip(metrics, values, limits), 1):
    status = "OK" if v < lim else "ALERT"
    print(f"{i}) {m}: {v}% (limit {lim}%) — {status}")

Нумерация для читаемого лога и сразу три колонки в цикле.

1. enumerate экономит строчки и бережёт от ошибок с «ручным» счётчиком.
2. zip позволяет синхронно итерировать сразу несколько коллекций.

Перешлите коллегам, пусть тоже прокачают свой Python-код! 😉
#python #codehacks #itertools

#SQLWednesday
Привет, сейчас середина недели а значит время SQL и сегодня рзберем задачку с собеседования ❓

Задача звучит так:
Найти топ-3(N) продаж (Top N Records per Group) для каждого продукта.
Дано табличка Sales:

| sale_id | product_id | sale_amount |
|---------|------------|-------------|
| S1      | P1         | 500         |
| S2      | P1         | 800         |
| S3      | P1         | 300         |
| S4      | P2         | 200         |
| S5      | P2         | 700         |
| S6      | P2         | 600         |

Нужно вывести для каждого product_id три записи с наибольшей sale_amount. Как это сделать ? Конечно с помощью оконной функции ROW_NUMBER()
1. Сначала делаем запрос с оконкой чтобы получить номера.

  SELECT
    product_id,
    sale_id,
    sale_amount,
    ROW_NUMBER() OVER (
      PARTITION BY product_id
      ORDER BY sale_amount DESC
    ) AS rn
  FROM Sales

- ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY sale_amount DESC)
- Партиционируем по продукту и сортируем от большего к меньшему сумму так мы получим по каждосу продукту номер в зависимости от суммы продажи.

2. Делаем завпрос из верхнего подзапроса, фильтруя результат.

SELECT
  t.product_id,
  t.sale_id,
  t.sale_amount
FROM (
  SELECT
    product_id,
    sale_id,
    sale_amount,
    ROW_NUMBER() OVER (
      PARTITION BY product_id
      ORDER BY sale_amount DESC
    ) AS rn
  FROM Sales
) AS t
WHERE t.rn <= 3;

- В подзапросе уже ROW_NUMBER() присваивает уникальный порядковый номер каждой строке внутри группы product_id, отсортированной по убыванию sale_amount.
Результат:

| product_id | sale_id | sale_amount |
|------------|---------|-------------|
| P1         | S2      | 800         |
| P1         | S1      | 500         |
| P1         | S3      | 300         |
| P2         | S5      | 700         |
| P2         | S6      | 600         |
| P2         | S4      | 200         |

Оконные функции вещь мощная, используйте ее для решения подобных задач!
Сохраняйте себе в закладки, пересылайте друзьям и делитесь своими альтернативными решениями в комментариях! 🙂
#DataEngineer #SQL #Interview #DataJungle

Испытание для Вас:

Добавим к нашей таблице `Sales` столбец `sale_date`:

| sale_id | product_id | sale_amount | sale_date   |
|---------|------------|-------------|-------------|
| S1      | P1         | 500         | 2024-06-01  |
| S2      | P1         | 800         | 2024-06-05  |
| S3      | P1         | 300         | 2024-07-02  |
| S4      | P2         | 200         | 2024-06-10  |
| S5      | P2         | 700         | 2024-06-15  |
| S6      | P2         | 600         | 2024-07-01  |

Задача:
Для каждого product_id и каждого месяца (`sale_date`) вывести топ-3 продаж по sale_amount.

💬 Поделитесь своим SQL-запросом в комментариях!

#PyTrickMonday - уже не понедельник ну и что вы мне сделаете 🙂

С началом новой недели, друзья! 🚀
Сегодня разбираем тему, которая реально прокачает читаемость вашего кода - аннотации типов и модуль #typing.

def greet(name: str, times: int) -> None:
    for _ in range(times):
        print(f"Привет, {name}!")

Зачем это нужно?
• 📈 IDE и линтеры (mypy) подсказывают ошибки ещё до запуска
• 🔍 Код самодокументируется - сразу видно, что функция ждёт и что возвращает
• 🤝 В командной разработке удобно понимать контракты без лишних комментариев

В примере выше вы видите достатчно простой пример аннотации типов, давайте рассмотрим более сложный вариант.

from typing import List, Dict

prices: List[float] = [123.4, 56.7, 89.0]
user_balances: Dict[str, float] = {"Alice": 1000.0, "Bob": 750.5}

- List[float] вместо list — защита от случайных вставок строк

Tuple и Union для сложных структур

from typing import Tuple, Union

def get_price(symbol: str) -> Union[float, None]:
    # вернёт цену или None, если не найдено

- Tuple[int, str] для жёсткого набора типов
- Union[X, Y] когда возможно несколько вариантов

TypedDict для диктов с фиксированным набором полей

from typing import TypedDict

class Order(TypedDict):
    id: int
    symbol: str
    amount: float

order: Order = {"id": 1, "symbol": "BTCUSDT", "amount": 0.5}

- Помогает инструментам подсказывать поля и их типы. На самом деле вместо TypedDict можно использовать dataclass, рассказал как вот в этом посте.

Generic-функции и Protocols (продвинутый уровень)

from typing import TypeVar, Iterable, List

T = TypeVar("T")

def to_list(iterable: Iterable[T]) -> List[T]:
    return list(iterable)

- TypeVar делает функцию универсальной для любых типов.

Выводы:
- Аннотации приводят к более понятному и безопасному коду.
- Инструменты проверки (mypy) ловят ошибки раньше, чем ваш код упадёт.
- В командной работе это просто must-have!

Перешлите коллегам, пусть тоже прокачают свой Python-код аннотациями типов и вы перестанете гадать что же возвращает функция 😉
#python #typing #mypy #codehacks #cleanCode