🌦️ Асинхронное взаимодействие с API OpenWeather

import aiohttp
import asyncio
from typing import Dict, Any

API_KEY = 'your_openweather_api_key'
BASE_URL = 'http://api.openweathermap.org/data/2.5/weather'

async def fetch_weather_data(city: str) -> Dict:
    async with aiohttp.ClientSession() as session:
        params = {'q': city, 'appid': API_KEY, 'units': 'metric'}
        async with session.get(BASE_URL, params=params) as response:
            if response.status == 200:
                return await response.json()
            else:
                raise Exception(f"Ошибка при получении данных: {response.status}")

async def main():
    city = "Moscow"
    try:
        weather_data = await fetch_weather_data(city)
        print(f"Погода в {city}: {weather_data['main']['temp']}°C")
    except Exception as e:
        print(f"Ошибка: {e}")

if __name__ == "__main__":
    asyncio.run(main())

📌 Этот код демонстрирует, как использовать асинхронные запросы для взаимодействия с OpenWeather API. Мы создаем асинхронный
клиент, который получает и обрабатывает данные о погоде для указанного города. Такой подход позволяет эффективно работать с API,
минимизируя задержки и улучшая производительность приложения.

Подпишись 👉🏻 @KodduuPython 🤖

📊 Автоматизация обработки данных в Pandas

import pandas as pd

def clean_data(file_path: str) -> pd.DataFrame:
    try:
        df = pd.read_csv(file_path)
        df.dropna(inplace=True)
        df.drop_duplicates(inplace=True)
        df.columns = df.columns.str.strip().str.lower().str.replace(' ', '_')
        return df
    except FileNotFoundError:
        print(f"Файл не найден: {file_path}")
        return pd.DataFrame()
    except pd.errors.EmptyDataError:
        print("Файл пустой.")
        return pd.DataFrame()
    except Exception as e:
        print(f"Ошибка при обработке файла: {e}")
        return pd.DataFrame()

# Пример использования
dataframe = clean_data('data.csv')
print(dataframe.head())

📌 Этот код автоматически очищает данные в CSV-файле: удаляет пропуски и дубликаты, а также форматирует имена столбцов. Это упрощает предварительную
обработку данных для анализа, что полезно аналитикам и разработчикам, работающим с большими наборами данных.

🛠 pip install pandas

Подпишись 👉🏻 @KodduuPython 🤖

Агенты общаются как люди (без оркестрации)

Вот пример общения агентов без оркестрации — они как люди сами решают, что делать и когда делать, а когда не делать...

1️⃣ Пришел запрос от пользователя.

2️⃣ Проверка пользователя на анонимность (если залогинен, то разрешаем).

3️⃣ Агент не блокировал НЕ персональный вопрос.

4️⃣ Агент-юрист понял, что вопрос не по его теме.

5️⃣ Агенту HR разрешено всё, он попытался найти инфу в документах компании, но не смог 😀.

6️⃣ Forward-agent понял, что на вопрос никто не ответил, и приказал его отправить людям 👨.

7️⃣ Агент task tracker взял и создал задачу на человека с этим вопросом.

Подпишись 👉🏻 @aigentto 🤖

🔧 Автоматизация управления облачными ресурсами с использованием AWS SDK и Python

import boto3
from botocore.exceptions import NoCredentialsError, PartialCredentialsError

def create_ec2_instance(image_id: str, instance_type: str, key_name: str) -> str:
    """Создает EC2-инстанс с указанными параметрами."""
    ec2 = boto3.resource('ec2')
    try:
        instance = ec2.create_instances(
            ImageId=image_id,
            InstanceType=instance_type,
            KeyName=key_name,
            MinCount=1,
            MaxCount=1
        )[0]
        return f"EC2 Instance {instance.id} создан."
    except (NoCredentialsError, PartialCredentialsError):
        return "Ошибка: Учетные данные AWS не найдены."
    except Exception as e:
        return f"Ошибка при создании инстанса: {e}"

# Пример использования
print(create_ec2_instance('ami-12345678', 't2.micro', 'my-key-pair'))

📌 Этот код демонстрирует автоматизацию управления облачными ресурсами AWS с использованием AWS SDK для Python (boto3). Скрипт позволяет создавать EC2-инстансы, что может значительно ускорить работу с ресурсами AWS для разработчиков и системных администраторов.

Подпишись 👉🏻 @KodduuPython 🤖

🔄 Автоматизация обновления данных в базах данных с использованием SQLAlchemy и Python

from sqlalchemy import create_engine, Column, Integer, String, Float, update
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import requests

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    price = Column(Float)

def fetch_product_data(api_url: str):
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Ошибка при запросе API: {e}")
        return []

def update_database(products_data):
    engine = create_engine('sqlite:///products.db')
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    
    for product in products_data:
        stmt = update(Product).where(Product.id == product['id']).values(price=product['price'])
        session.execute(stmt)
    
    session.commit()
    session.close()

# Пример использования
product_api_url = "https://api.example.com/products"
products = fetch_product_data(product_api_url)
if products:
    update_database(products)

📌 Этот код демонстрирует, как использовать SQLAlchemy и Python для автоматизации обновления данных в базах данных.

Скрипт подключается к базе, получает данные из внешнего API, и обновляет цены продуктов в базе данных.

Такой подход полезен для интеграции данных и их синхронизации между системами, что актуально для интернет-магазинов и аналитических платформ.

Подпишись 👉🏻 @KodduuPython 🤖

Вышла статья на habr по следам проблемы циклов 🧐

Подпишись 👉🏻 @aigentto 🤖

📄 Генерация отчетов в формате PDF с Python и ReportLab

from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
from reportlab.lib.units import inch

def create_pdf(filename: str):
    try:
        c = canvas.Canvas(filename, pagesize=A4)
        width, height = A4
        
        # Заголовок
        c.setFont("Helvetica-Bold", 16)
        c.drawString(1 * inch, height - 1 * inch, "Отчет о продажах")

        # Таблица данных
        data = [("Продукт", "Количество", "Цена"),
                ("Продукт A", "10", "$100"),
                ("Продукт B", "20", "$200"),
                ("Продукт C", "15", "$150")]

        c.setFont("Helvetica", 12)
        y_position = height - 1.5 * inch
        for row in data:
            x_position = 1 * inch
            for item in row:
                c.drawString(x_position, y_position, item)
                x_position += 2 * inch
            y_position -= 0.5 * inch

        # Завершение
        c.showPage()
        c.save()
        print(f"PDF создан: {filename}")
    except Exception as e:
        print(f"Ошибка при создании PDF: {e}")

# Пример использования
create_pdf("sales_report.pdf")

📌 Этот код демонстрирует базовый процесс создания PDF-документа с помощью библиотеки ReportLab. Скрипт генерирует отчет с заголовком и таблицей
данных, что полезно для автоматизации отчетности в бизнесе. ReportLab позволяет легко добавлять текст, таблицы и графики в документы, упрощая создание профессиональных отчетов.

🛠 pip install reportlab

Подпишись 👉🏻 @KodduuPython 🤖

📊 Автоматизация планирования задач с использованием Python и Apache Airflow

from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def my_task():
    print("Выполняется задача!")

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 10, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG('my_simple_dag', default_args=default_args, schedule_interval=timedelta(days=1)) as dag:
    start = DummyOperator(task_id='start')
    execute_task = PythonOperator(task_id='execute_task', python_callable=my_task)
    end = DummyOperator(task_id='end')

    start >> execute_task >> end

# Пример использования
# После настройки Apache Airflow, DAG будет доступен в веб-интерфейсе Airflow

📌 Этот код демонстрирует создание простого DAG (Directed Acyclic Graph) в Apache Airflow. DAG состоит из трёх задач: стартовой, выполнения
Python-функции и завершения. Airflow автоматически планирует выполнение этих задач по расписанию, что идеально подходит для автоматизации процессов обработки данных и бизнес-аналитики.

Подпишись 👉🏻 @KodduuPython 🤖

Не забываем про Самый быстрый курс по Python 🧐

Сейчас как раз идет какая-то титаническая распродажа на Stepik.

Покупаем себе, друзьям и даже за деньги компании можно купить🔥🔥🔥

Подпишись 👉🏻 @KodduuPython 🤖

🔍 Поиск файла с помощью рекурсивного обхода

import os

def find_file(filename: str, search_path: str) -> str:
    try:
        for root, dirs, files in os.walk(search_path):
            if filename in files:
                return os.path.join(root, filename)
        raise FileNotFoundError(f"'{filename}' не найден в '{search_path}'")
    except Exception as e:
        return str(e)

# Пример использования
file_path = find_file('example.txt', '/path/to/search')
print(file_path)

📌 Эта функция позволяет рекурсивно искать файл в указанной директории и её поддиректориях. Она использует os.walk для
обхода файловой системы и возвращает полный путь к файлу, если он найден. Полезно для автоматизации задач, связанных с
управлением файлами, и для скриптов, требующих динамического поиска файлов в системе.

Подпишись 👉🏻 @KodduuPython 🤖

Добро пожаловать в no-code решения

Не будут эти решения работать в реальных боевых сценариях никогда. Это уже 4-я итерация с 00-х годов по разным темам, сейчас ИИ.

Но как же заманчиво это для клиента. Сам не разбираюсь, нанимать разработчиков не буду.

Надо понимать наивность такого подхода — я тоже сам могу построить себе дом, но я чётко осознаю: либо я потрачу много лет своего времени, разберусь и стану строителем — нафига мне это надо?

Либо я заплачу не временем, а деньгами тем, кто уже разбирается (строителям), и продать мне дом типа n8n по схеме «собери сам» не получится 🤷

Подпишись 👉🏻 @aigentto 🤖

🔄 Сравнение многопоточности и многопроцессорности в Python: threading vs. multiprocessing

import threading
import time

def io_task():
    """Симуляция задачи, связанной с вводом-выводом."""
    print(f"Начало I/O задачи в потоке {threading.current_thread().name}")
    time.sleep(2)
    print(f"Завершение I/O задачи в потоке {threading.current_thread().name}")

# Пример многопоточности для I/O bound задач
threads = 
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

📌 В этом коде демонстрируется использование многопоточности в Python для выполнения задач, связанных с вводом-выводом.
Модуль threading позволяет параллельно выполнять несколько потоков, что ускоряет операции ввода-вывода. Это полезно для
оптимизации приложений, где много времени занимает ожидание завершения операций ввода-вывода.

Подпишись 👉🏻 @KodduuPython 🤖

🌀 Оптимизация памяти с генераторами и yield

def read_large_file(file_path: str):
    """Читает файл построчно, используя генератор."""
    with open(file_path, 'r', encoding='utf-8') as file:
        for line in file:
            yield line.strip()

def process_lines(file_path: str):
    """Обрабатывает строки большого файла."""
    try:
        for line in read_large_file(file_path):
            print(f"Обработка строки: {line[:50]}...")
    except FileNotFoundError:
        print(f"Файл {file_path} не найден.")
    except Exception as e:
        print(f"Произошла ошибка: {e}")

# Пример использования
process_lines('large_file.txt')

📌 Этот код демонстрирует, как генераторы позволяют эффективно обрабатывать большие файлы, не загружая их полностью в
память. Функция read_large_file использует yield для построчного чтения файла, что особенно полезно при работе с
объёмными данными. Такой подход оптимизирует использование памяти и улучшает производительность программы.

Подпишись 👉🏻 @KodduuPython 🤖

🔄 Эффективное использование многопоточности и асинхронности: threading vs. asyncio

import asyncio
import threading
import requests
from time import time

def fetch_data_with_threads(urls):
    """Загрузка данных с помощью многопоточности."""
    def fetch(url):
        response = requests.get(url)
        print(f"Получены данные с {url}: {len(response.content)} байт")

    threads = 
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

async def fetch_data_with_asyncio(urls):
    """Загрузка данных с помощью асинхронности."""
    async def fetch(url):
        loop = asyncio.get_event_loop()
        response = await loop.run_in_executor(None, requests.get, url)
        print(f"Получены данные с {url}: {len(response.content)} байт")

    tasks = 
    await asyncio.gather(*tasks)

# Пример использования
urls = ['https://example.com'] * 5

# Многопоточность
start = time()
fetch_data_with_threads(urls)
print(f"Многопоточность: {time() - start:.2f} секунд")

# Асинхронность
start = time()
asyncio.run(fetch_data_with_asyncio(urls))
print(f"Асинхронность: {time() - start:.2f} секунд")

📌 Этот код показывает, как можно загружать данные с нескольких URL одновременно, используя многопоточность и асинхронное
программирование в Python. Модуль threading позволяет запускать несколько потоков для выполнения сетевых запросов, в то
время как asyncio обеспечивает асинхронное выполнение, что может быть более эффективно для задач, связанных с
вводом-выводом. Оба подхода полезны для повышения производительности приложений, работающих с сетью.

Подпишись 👉🏻 @KodduuPython 🤖

Ускорение Python кода с использованием Numba

import numpy as np
from numba import jit

# Пример на Numba
@jit(nopython=True)
def sum_of_squares(arr):
    total = 0
    for i in arr:
        total += i * i
    return total

# Пример использования
arr = np.arange(1e6)

# Используем Numba
print("Numba: ", sum_of_squares(arr))

📌 Этот код демонстрирует использование библиотеки Numba для ускорения вычислений в Python. Numba компилирует функции на
лету, что позволяет быстро улучшить производительность без дополнительных шагов. Этот инструмент особенно полезен для
вычислительно интенсивных задач, таких как обработка массивов или научные вычисления.

Подпишись 👉🏻 @KodduuPython 🤖

Люди подстроятся под ботов

Любые технологии на самом деле немного ухудшают пользовательский опыт. Например, когда все перешли на цифровую связь (сотовые, IP телефония, видеоконференции), то автоматически получили задержку связи до 100 мс. Это дико много по сравнению с почти скоростью света, которая была доступна на проводных телефонах, соединённых через АТС.

Но все привыкли. Мы теперь делаем паузу после завершения своего монолога и немного ждём по завершению чужого. Попробуйте сейчас поговорить по проводному телефону — вы офигеете, насколько это удобнее ☎️

То же самое будет с тупыми чат-ботами, которые сейчас все навнедряли и которые всех бесят. Вы всё привыкнете. Вы научитесь говорить так и использовать такие запросы, которые с большей вероятностью будут понятны боту. Люди адаптивны, поэтому подстроятся под ботов 🙇

Подпишись 👉🏻 @aigentto 🤖

🔍 Кластеризация данных с K-средними

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.datasets import make_blobs
import numpy as np

def prepare_data(n_samples: int = 300, n_features: int = 2, centers: int = 3) -> np.ndarray:
    # Создание случайных данных с центрами
    X, _ = make_blobs(n_samples=n_samples, n_features=n_features, centers=centers, random_state=42)
    return X

def cluster_data(X: np.ndarray, n_clusters: int = 3) -> np.ndarray:
    # Масштабирование данных
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(X)

    # Кластеризация методом K-средних
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    kmeans.fit(X_scaled)
    return kmeans.labels_

# Пример использования
if __name__ == "__main__":
    data = prepare_data()
    labels = cluster_data(data)
    print("Кластеры:", labels)

📌 Этот код иллюстрирует использование алгоритма K-средних для кластеризации данных. Мы создаём случайные данные, масштабируем их и применяем K-средние для разделения на кластеры. Такой подход может быть полезен в маркетинге для сегментации клиентов или в биологии для классификации видов.

Подпишись 👉🏻 @KodduuPython 🤖

🔥 Распродажа на Stepik закончится 4 декабря в 23:59 🏃🏃‍♀️

👉 Самое время пройти Python: самый быстрый курс 🧐

👉 Или окунутся в мир Data Science на курсе Python Data Science: самый быстрый курс 🤓

👉 Быстро выучить JS на курсу JavaScript: самый быстрый курс 😎

👉 И наконец узнать все сразу в программе курсов Junior Python Developer и Data Scientist +интервью тест 🧑‍🎓

КУПИТЬ КУРСЫ МОЖНО ЗА СЧЕТ РАБОТАДАТЕЛЯ 📌📌📌

Подпишись 👉🏻 @KodduuPython 🤖

🚀 Поиск кратчайшего пути с использованием алгоритма A*

from heapq import heappop, heappush

def heuristic(a, b):
    return abs(a[0] - b[0]) + abs(a[1] - b[1])

def a_star_search(graph, start, goal):
    open_set = []
    heappush(open_set, (0, start))
    came_from = {}
    g_score = {start: 0}
    f_score = {start: heuristic(start, goal)}

    while open_set:
        _, current = heappop(open_set)

        if current == goal:
            return reconstruct_path(came_from, current)

        for neighbor in graph.get(current, []):
            tentative_g_score = g_score + graph
            if tentative_g_score < g_score.get(neighbor, float('inf')):
                came_from = current
                g_score = tentative_g_score
                f_score = tentative_g_score + heuristic(neighbor, goal)
                if neighbor not in open_set:
                    heappush(open_set, (f_score, neighbor))

    return None

def reconstruct_path(came_from, current):
    total_path = 
    while current in came_from:
        current = came_from
        total_path.append(current)
    return total_path[::-1]

# Пример использования
graph = {
    (0, 0): {(1, 0): 1, (0, 1): 1},
    (1, 0): {(0, 0): 1, (1, 1): 1, (2, 0): 1},
    (0, 1): {(0, 0): 1, (1, 1): 1},
    (1, 1): {(1, 0): 1, (0, 1): 1, (2, 1): 1},
    (2, 0): {(1, 0): 1, (2, 1): 1},
    (2, 1): {(2, 0): 1, (1, 1): 1}
}

start = (0, 0)
goal = (2, 1)
path = a_star_search(graph, start, goal)
print("Кратчайший путь:", path)

📌 Этот код демонстрирует реализацию алгоритма A* для поиска кратчайшего пути на графе. Он использует эвристическую функцию для оценки наилучшего маршрута и поддерживает открытый список с приоритетом для исследования. Такой подход может быть полезен в приложениях навигации и разработке игр для поиска оптимальных путей.

Подпишись 👉🏻 @KodduuPython 🤖