DataДжунгли🌳
286 subscribers
11 photos
8 videos
1 file
27 links
Data Engineering на пальцах:
🥸Про Airflow-DAG.
🍀ETL, с сарказмом.
🅾️оптимизируем SQL.
🐍Python трюки каждый пн.
Download Telegram
#PyTrickMonday

dict.get() : маленькая таблетка от KeyError-ов

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


payload = {"user": {"name": "Alice"}}

# Явное обращение по ключу
nickname = payload["user"]["nickname"] # KeyError, DAG падает

# Спокойный вариант c .get()
nickname = payload.get("user", {}).get("nickname", "Anon")

# "Если ты сигма" - способ (когда поле может отсутствовать целиком)
nickname = (payload.get("user") or {}).get("nickname", "Anon")


Так вот я вам предлагаю использовать .get() ВСЕГДА и у меня на это пять три причины:

1️⃣Безопасность: API вернул 200 OK, но ключа "user" нет - прямое обращение бросит KeyError и уронит скрипт. .get() вернёт None (или ваш дефолт) и программа продолжит работу.

2️⃣Меньше грязного кода: вместо громоздкого: (Плоское лучше вложенного ZenOfPython)

if "key" in resp:
value = resp["key"]

# одна строчка
value = resp.get("key")

3️⃣Гибкий дефолт: jobs.get("DA", "DE") # вернёт 'DE', если ключа 'DA' нет

💡 Если надо "нырнуть" глубоко и не облажаться по пути, так как .get есть только у словарей, можно дефлтом указывать пустой словарь и ваш код не упадет.
email = payload.get("user", {}).get("contacts", {}).get("email")

📣 Отправь этот полезный пост коллеге чтобы он/она не ловили KeyError и были тебе всегда благодарны 😎

#DETip #Python #DataJungle
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥11
Media is too big
VIEW IN TELEGRAM
Ну что сегодня пятница а значит время продолжить большой разбор #Airflow 🚀

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

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


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

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

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

В следующем видео сделаем #CICD и установим #Postgres на наш сервер чтобы там хранить данные не пропускайте! 🙄
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥7
#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 подскажет нужные атрибуты и защитит от опечаток.

Пересылайте коллегам чтобы они уже сегодня использовали 👀
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥6😱2
#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
Please open Telegram to view this post
VIEW IN TELEGRAM
👍6
#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
Please open Telegram to view this post
VIEW IN TELEGRAM
👍7
#PyTrickMonday
any() и all() — твои мини-фильтры
или как проверять условия в коллекциях без циклов и лишнего кода 🔍

Часто в коде хочется спросить у Python:
“Есть ли хоть один элемент, который подходит под условие?”
“А все ли элементы ок?”
И вот тут any() и all() — настоящие ниндзя 🥷

nums = [2, 4, 6, 8]

# Есть ли хоть одно нечётное? (any)
has_odd = any(n % 2 for n in nums) # False

# Все ли чётные? (all)
all_even = all(n % 2 == 0 for n in nums) # True


💡 Почему это удобно:
1️⃣ Читаемость: any(...) читается как хоть один, all(...) - все.
2️⃣ Меньше кода: вместо громоздкого for/if/break - одна строка.
3️⃣ Эффективность: останавливаются при первом подходящем/неподходящем элементе.
4️⃣ Гибкость: можно проверять не только числа, но и строки, объекты, словари.

Ещё пример из жизни:


emails = ["[email protected]", "", "[email protected]"]

# Проверить, что у всех сотрудников есть email
if all(emails):
print("Все указали адреса")
else:
print("Кто-то забыл email")

# Проверить, что хотя бы один email корпоративный
if any(e.endswith("@company.com") for e in emails if e):
print("Есть корпоративный адрес")


📌 Запомни:
any([]) → False (в пустом списке нет ни одного истинного элемента)
all([]) → True (математика: вакуумно-истинно 😏)

Отправь этот пост коллеге- пусть тоже перестанет писать километры циклов!
#Python #DataJungle
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥4
#PyTrickMonday

Часть 1. Базовое знакомство с асинхроном.

asyncio - когда Python становится акробатом 🤸‍♂️
или как запускать кучу задач одновременно, не дожидаясь поочерёдно

Глупый пример но тем не менее 🙂
"тебе надо сходить за хлебом, оплатить коммуналку и приготовить кофе."
Синхронно: делаешь всё подряд, ждёшь пока закипит вода , потом топаешь в магазин.
Асинхронно: ставишь чайник, пока он греется — бежишь за хлебом, по пути платишь счета.

так работает asyncio

import asyncio

async def task(name, delay):
await asyncio.sleep(delay)
print(f"{name} done after {delay}s")

async def main():
await asyncio.gather(
task("💡 Coffee", 3),
task("🍞 Bread", 2),
task("💸 Bills", 1),
)

asyncio.run(main())

Результат: всё завершается параллельно, а не по очереди.

• async — объявляешь асинхронную функцию;
• await — говоришь “подожди, но пока можно делать другое”;
• gather — запускает несколько задач одновременно.

📌 Когда использовать asyncio:
Когда много I/O-операций: запросы к API, базы данных, парсинг сайтов.
Когда важна скорость реакции: чат-боты, стриминг, веб-сервисы.
Когда нужно уместить сотни/тысячи соединений в один процесс.

🚫 Когда НЕ использовать:
Для тяжёлых вычислений (ML, обработка больших массивов) — тут поможет multiprocessing или Spark.
Если у тебя простая линейная программа (асинхронщина только усложнит код).
Когда надо жать CPU на 100% (асинхронный код от этого не ускорится).

В следующей части покажу, как управлять этими задачами: ставить таймауты, отменять и ловить «медленников».

#Python #Asyncio #DataJungle
2😱2
#PyTrickMonday
asyncio — Часть 2
Контроль задач: таймауты, отмена, обработка «медленников» 🎪

Когда задач много, «запустить и забыть» уже не работает. Нужен контроль: кто отвалился, кто завис, кого пора отменить.

1) gather vs wait vs as_completed
asyncio.gather(*tasks) — простейший «всё сразу».
Удобно, когда:
• все задачи равнозначны;
• результат нужен только когда всё готово.
• Плюс: может собирать результаты в порядке аргументов.
• Минус: по умолчанию ошибка одной задачи «роняет» всё (см. return_exceptions=True).

asyncio.wait(tasks, return_when=...) — когда нужен статус-контроль:
• FIRST_COMPLETED — продолжай, как только что-то завершилось;
• FIRST_EXCEPTION — реагируй на первую ошибку;
• ALL_COMPLETED — эквивалент дождаться всех, но с наборами done/pending.
• asyncio.as_completed(tasks) — стримишь результаты по мере готовности:
• обрабатываешь «быстрые» ответы, не дожидаясь «черепах».

2) Таймауты: жёсткий дедлайн для всего или для каждой задачи
Глобальный таймаут (для группы)



import asyncio
async def api_call(t): # имитируем сетевой вызов
await asyncio.sleep(t)
return f"ok in {t}s"

async def main():
tasks = [api_call(1), api_call(5), api_call(10)]
try:
# «Всё должно уложиться за 6 секунд»
res = await asyncio.wait_for(asyncio.gather(*tasks), timeout=6)
print(res)
except asyncio.TimeoutError:
print(" Глобальный таймаут: не все успели!")

asyncio.run(main())



- Индивидуальные таймауты (каждой задаче свой лимит)


import asyncio

async def fetch(t):
await asyncio.sleep(t)
return t

async def safe_fetch(t, limit=2):
try:
return await asyncio.wait_for(fetch(t), timeout=limit)
except asyncio.TimeoutError:
return f"timeout>{limit}s"

async def main():
tasks = [safe_fetch(1), safe_fetch(3), safe_fetch(0.5)]
print(await asyncio.gather(*tasks))

asyncio.run(main())


3) Отмена задач: .cancel() и обработка CancelledError

Задачи нужно уметь «выключать культурно», освобождая ресурсы.


import asyncio

async def worker():
try:
while True:
await asyncio.sleep(0.5) # работа
except asyncio.CancelledError:
print("🔌 Получил cancel, закрываю соединения...")
# тут — закрытие сессий/курсов/файлов
raise # пробрасываем дальше (правильно!)

async def main():
task = asyncio.create_task(worker())
await asyncio.sleep(2) # пусть поработает
task.cancel()
try:
await task
except asyncio.CancelledError:
print(" Отмена завершена корректно")

asyncio.run(main())


4) Обрабатывать результаты «как приходят» — as_completed
Отлично подходит для «медленного зоопарка» API.


import asyncio
from random import randint

async def call(i):
t = randint(1, 5)
await asyncio.sleep(t)
return f"#{i} in {t}s"

async def main():
tasks = [asyncio.create_task(call(i)) for i in range(5)]
for fut in asyncio.as_completed(tasks):
try:
res = await fut
print("", res)
except Exception as e:
print(" ошибка:", e)

asyncio.run(main())


5) Частые грабли и лайфхаки

• gather(..., return_exceptions=True) — не роняй весь батч из-за одной ошибки; обработай её в результате.
• Создавай задачи через asyncio.create_task(), если нужно запустить и продолжать работу (иначе просто await подряд).
• Не блокируй event loop: никакого time.sleep() и долгих CPU-циклов внутри async — только await I/O или выноси CPU в concurrent.futures/multiprocessing.
• Таймауты — не панацея: всегда закрывай ресурсы и корректно обрабатывай отмену.
• Смешивать threading и asyncio можно, но осторожно: для блокирующих библиотек используй loop.run_in_executor(...).


Контролируй асинхронность - и она начнёт экономить тебе время, а не нервы 😏 делитесь этим с друзьями и коллегами !
#Python #Asyncio #DataJungle
👍32