🔍 Эффективность использования Generators vs. List Comprehensions в Python

def list_comprehension_example(n: int):
    return 

def generator_example(n: int):
    return (x * x for x in range(n))

# Пример использования
if __name__ == "__main__":
    n = 10**6

    # Использование спискового включения
    squares_list = list_comprehension_example(n)
    print(f"Сгенерирован список квадратов длиной {len(squares_list)}")

    # Использование генератора
    squares_generator = generator_example(n)
    print("Сгенерирован генератор квадратов")

    # Пример итерации по генератору
    sum_of_squares = sum(squares_generator)
    print(f"Сумма квадратов до {n}: {sum_of_squares}")

📌 Этот код демонстрирует два подхода к созданию последовательностей в Python: списковые включения и генераторы.

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

Генераторы полезны при работе с большими данными, когда важно снижать нагрузку на оперативную память.

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

🔄 Асинхронное программирование vs. многопоточность в Python

import asyncio
import concurrent.futures
import requests

async def fetch_url_async(url: str) -> str:
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        response = await loop.run_in_executor(pool, requests.get, url)
        return response.text

def fetch_url_sync(url: str) -> str:
    response = requests.get(url)
    return response.text

async def main():
    url = "https://www.example.com"
    
    # Асинхронный вызов
    async_result = await fetch_url_async(url)
    print("Асинхронный результат получен")

    # Синхронный вызов
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future = executor.submit(fetch_url_sync, url)
        sync_result = future.result()
        print("Синхронный результат получен")

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

📌 Этот код демонстрирует различия между асинхронным программированием и многопоточностью в Python для выполнения сетевых запросов.

Асинхронная функция fetch_url_async использует asyncio и ThreadPoolExecutor для выполнения запросов без блокировки основного потока,
в то время как синхронная функция fetch_url_sync выполняет их в отдельном потоке.

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

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

💡 Генераторы vs. Списочные включения: Эффективная обработка данных

def process_large_file(file_path: str) -> int:
    """Считает количество строк, содержащих слово 'Python' в большом файле."""
    count = 0
    try:
        with open(file_path, 'r', encoding='utf-8') as file:
            for line in file:
                if 'Python' in line:
                    count += 1
    except FileNotFoundError:
        print(f"Файл {file_path} не найден.")
    except Exception as error:
        print(f"Произошла ошибка: {error}")
    return count

# Пример использования
file_path = 'large_text_file.txt'
python_count = process_large_file(file_path)
print(f"Количество строк с 'Python': {python_count}")

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

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

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

🛠️ Автоматизация тестирования API с Pytest и Requests

import requests
import pytest

BASE_URL = "https://jsonplaceholder.typicode.com"

def test_get_post():
    response = requests.get(f"{BASE_URL}/posts/1")
    assert response.status_code == 200
    assert response.json()["id"] == 1

def test_create_post():
    new_post = {"title": "foo", "body": "bar", "userId": 1}
    response = requests.post(f"{BASE_URL}/posts", json=new_post)
    assert response.status_code == 201
    assert response.json()["title"] == new_post["title"]

def test_update_post():
    updated_data = {"title": "foobar"}
    response = requests.put(f"{BASE_URL}/posts/1", json=updated_data)
    assert response.status_code == 200
    assert response.json()["title"] == updated_data["title"]

def test_delete_post():
    response = requests.delete(f"{BASE_URL}/posts/1")
    assert response.status_code == 200

if __name__ == "__main__":
    pytest.main()

📌 Этот код демонстрирует, как использовать Pytest и Requests для автоматизации тестирования API. Он включает тесты для операций GET, POST, PUT и DELETE, что позволяет убедиться в корректной работе вашего API. Эти тесты могут быть интегрированы в CI/CD процессы, чтобы автоматически проверять API при каждом обновлении кода.

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

🔍 Автоматизированная система мониторинга веб-сайтов с BeautifulSoup и Selenium

from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
import requests
import time

def get_page_content(url: str) -> str:
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response.text
    except requests.RequestException as e:
        print(f"Ошибка при получении содержимого страницы: {e}")
        return ""

def parse_content(html: str) -> str:
    soup = BeautifulSoup(html, 'html.parser')
    title = soup.find('title').text if soup.find('title') else 'Заголовок не найден'
    return title

def monitor_website(url: str, check_interval: int):
    content = get_page_content(url)
    current_title = parse_content(content)
    print(f"Начальный заголовок: {current_title}")

    options = Options()
    options.headless = True
    service = Service('/path/to/chromedriver')
    driver = webdriver.Chrome(service=service, options=options)
    driver.get(url)

    try:
        while True:
            driver.refresh()
            new_content = driver.page_source
            new_title = parse_content(new_content)
            if new_title != current_title:
                print(f"Содержание сайта изменилось! Новый заголовок: {new_title}")
                current_title = new_title
            time.sleep(check_interval)
    except Exception as e:
        print(f"Ошибка во время мониторинга: {e}")
    finally:
        driver.quit()

# Пример использования
monitor_website("https://example.com", check_interval=60)

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

🛠 pip install beautifulsoup4 selenium requests

💡 Не забудьте скачать и указать путь к chromedriver для корректной работы Selenium.

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

📄 Автоматизация обработки документов с использованием PyPDF2

import PyPDF2

def merge_pdfs(pdf_list, output_path):
    pdf_writer = PyPDF2.PdfWriter()
    
    for pdf_file in pdf_list:
        try:
            with open(pdf_file, 'rb') as file:
                pdf_reader = PyPDF2.PdfReader(file)
                for page in range(len(pdf_reader.pages)):
                    pdf_writer.add_page(pdf_reader.pages)
        except Exception as e:
            print(f"Ошибка при чтении {pdf_file}: {e}")
    
    try:
        with open(output_path, 'wb') as output_pdf:
            pdf_writer.write(output_pdf)
        print(f"Объединенный PDF сохранен как {output_path}")
    except Exception as e:
        print(f"Ошибка при записи объединенного PDF: {e}")

# Пример использования
pdf_files = ["document1.pdf", "document2.pdf"]
merge_pdfs(pdf_files, "merged_document.pdf")

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

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

🕸️ Автоматизация сбора данных с веб-форм с помощью Selenium

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
import time

def setup_driver():
    chrome_options = Options()
    chrome_options.add_argument("--headless")  # Запуск браузера в режиме без интерфейса
    service = Service(executable_path='path/to/chromedriver')
    driver = webdriver.Chrome(service=service, options=chrome_options)
    return driver

def fill_form_and_submit(url, form_data):
    driver = setup_driver()
    try:
        driver.get(url)
        for field_id, value in form_data.items():
            element = driver.find_element(By.ID, field_id)
            element.clear()  # Очистка существующего текста в поле
            element.send_keys(value)  # Ввод нового текста
        
        submit_button = driver.find_element(By.ID, "submit")
        submit_button.click()
        
        time.sleep(3)  # Ожидание загрузки страницы
        return driver.page_source
    except Exception as e:
        print(f"Ошибка при отправке формы: {e}")
    finally:
        driver.quit()

# Пример использования
url = "http://example.com/form"
form_data = {
    "name": "John Doe",
    "email": "john.doe@example.com",
    "message": "Hello, this is a test message!"
}
response = fill_form_and_submit(url, form_data)
print(response[:500])  # Вывод первых 500 символов HTML страницы

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

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

👉 Ключевой бесплатный курс для разминки мозга 🤯 и понимания как работает компьютер 💻

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

📊 Управление задачами с Celery

from celery import Celery

# Создание приложения Celery
app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def send_email(recipient: str, subject: str, body: str) -> str:
    # Имитация отправки email
    print(f"Отправка email на {recipient} с темой '{subject}'")
    return "Email отправлен!"

# Пример использования
if __name__ == "__main__":
    result_email = send_email.delay("example@example.com", "Hello!", "Это тестовое сообщение.")
    
    print(result_email.get(timeout=10))

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

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

🔍 Автоматизация тестирования с Pytest

import pytest

# Функция для тестирования
def add(a: int, b: int) -> int:
    return a + b

# Тестовая функция
def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0

# Параметризация тестов
@pytest.mark.parametrize("a, b, expected", [
    (1, 1, 2),
    (-1, -1, -2),
    (100, 200, 300),
])
def test_add_parametrized(a, b, expected):
    assert add(a, b) == expected

# Запуск тестов командой: pytest имя_файла.py
📌 Данный код демонстрирует основы использования Pytest для автоматизации тестирования функций в Python. Он включает тесты для функции сложения и пример параметризации тестов для проверки различных входных данных. Это полезно для повышения качества вашего кода и обеспечения его стабильности при изменениях.

🛠 Для начала работы установите Pytest с помощью: pip install pytest.

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

🔍 Увеличение производительности с Dask

import dask.dataframe as dd
import pandas as pd

def process_large_csv(file_path: str, column: str) -> pd.Series:
    try:
        # Чтение большого CSV-файла с помощью Dask
        ddf = dd.read_csv(file_path)
        
        # Выполнение операции группировки и агрегации
        result = ddf.groupby(column).size().compute()
        
        return result
    except Exception as e:
        print(f"Ошибка обработки файла: {e}")
        return pd.Series()

# Пример использования
if __name__ == "__main__":
    file_path = 'large_data.csv'
    column = 'category'
    result = process_large_csv(file_path, column)
    print(result)

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

🛠 pip install dask pandas

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

Вышла новая статья на habr "Как извлечь ДНК автора? Креатив от LLM" 💡

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

🔍 Асинхронный веб-скрейпинг с Aiohttp и BeautifulSoup

import aiohttp
import asyncio
from bs4 import BeautifulSoup

async def fetch(session, url):
    async with session.get(url) as response:
        if response.status == 200:
            html = await response.text()
            # Парсинг HTML для извлечения заголовка
            soup = BeautifulSoup(html, 'html.parser')
            title = soup.title.string if soup.title else 'No Title Found'
            return title
        else:
            raise Exception(f"Ошибка при получении {url}: статус {response.status}")

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        titles = await asyncio.gather(*tasks, return_exceptions=True)
        return titles

# Пример использования
urls = ['https://www.python.org', 'https://www.example.com']
titles = asyncio.run(main(urls))
print(titles)

📌 Этот код выполняет асинхронный веб-скрейпинг, извлекая заголовки страниц с заданных URL. Используя aiohttp для асинхронных HTTP-запросов и BeautifulSoup для парсинга HTML, он эффективно обрабатывает множество страниц одновременно. Этот подход особенно полезен для обработки больших объемов данных без блокировки выполнения программы.

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

🖼️ Простая обработка изображений с PIL и OpenCV

from PIL import Image, ImageFilter
import cv2
import numpy as np

def process_image(image_path: str, output_path: str):
    try:
        # Открываем изображение с помощью PIL
        pil_image = Image.open(image_path)
        
        # Применяем размытие
        blurred_image = pil_image.filter(ImageFilter.BLUR)
        
        # Сохраняем результат
        blurred_image.save(output_path)
        
        # Чтение изображения с помощью OpenCV
        cv_image = cv2.imread(output_path)
        
        # Конвертируем в градации серого
        gray_image = cv2.cvtColor(cv_image, cv2.COLOR_BGR2GRAY)
        
        # Применяем детектор границ Canny
        edges = cv2.Canny(gray_image, 100, 200)
        
        # Сохраняем изображение с границами
        cv2.imwrite(output_path.replace('.jpg', '_edges.jpg'), edges)
        
    except Exception as e:
        print(f"Ошибка обработки изображения: {e}")

# Пример использования
process_image('input.jpg', 'output.jpg')

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

🛠 pip install pillow opencv-python-headless

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

🛠️ Быстрое RESTful API с FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
    id: int
    name: str
    description: Optional[str] = None  # Описание элемента

# Пример данных
items_db = [
    {"id": 1, "name": "item1", "description": "Описание первого элемента"},
    {"id": 2, "name": "item2", "description": "Описание второго элемента"},
]

@app.get("/items/", response_model=List[Item])
async def read_items():
    return items_db

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    for item in items_db:
        if item["id"] == item_id:
            return item
    raise HTTPException(status_code=404, detail="Элемент не найден")

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    for existing_item in items_db:
        if existing_item["id"] == item.id:
            raise HTTPException(status_code=400, detail="ID уже существует")
    items_db.append(item.dict())
    return item

# Пример использования:
# 1. Запустите сервер: uvicorn filename:app --reload
# 2. Посетите http://127.0.0.1:8000/items/ для просмотра всех элементов
📌 Этот код демонстрирует создание простого RESTful API с использованием FastAPI. API позволяет получать список элементов, искать элемент по ID и добавлять новые элементы с уникальными ID. FastAPI обеспечивает быструю разработку и безопасность API благодаря встроенной валидации и асинхронному выполнению.

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

📊 Простая агрегация данных с Pandas

import pandas as pd

def aggregate_sales_data(file_path: str) -> pd.DataFrame:
    try:
        data = pd.read_csv(file_path, parse_dates=['Date'])
        
        # Группировка данных по месяцам и суммирование продаж
        monthly_sales = (
            data.groupby(data['Date'].dt.to_period('M'))
            .agg(total_sales=('Sales', 'sum'))
            .reset_index()
        )
        return monthly_sales
    except FileNotFoundError:
        print("Файл не найден. Проверьте путь к файлу.")
        return pd.DataFrame()
    except Exception as e:
        print(f"Произошла ошибка: {e}")
        return pd.DataFrame()

# Пример использования
file_path = 'sales_data.csv'
monthly_sales = aggregate_sales_data(file_path)
print(monthly_sales)

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

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

* С новым годом *

import os
import time
import random

WIDTH = 80
HEIGHT = 20

# ASCII-art цифр (5 строк)
DIGITS = {
    "0": [
        " █████ ",
        " █   █ ",
        " █   █ ",
        " █   █ ",
        " █████ "
    ],
    "2": [
        " █████ ",
        "     █ ",
        " █████ ",
        " █     ",
        " █████ "
    ],
    "6": [
        " █████ ",
        " █     ",
        " █████ ",
        " █   █ ",
        " █████ "
    ]
}

# собрать строку 2026
def get_2026_art():
    art_lines = [""] * 5
    for digit in "2026":
        for i in range(5):
            art_lines[i] += DIGITS[digit][i] + "  "
    return art_lines

# создаем пустое поле для снега
def init_snow():
    return [[random.choice([False, True, False, False, False]) for _ in range(WIDTH)] for _ in range(HEIGHT)]

# обновление позиции снега
def update_snow(snow):
    new_snow = [[False]*WIDTH for _ in range(HEIGHT)]
    for y in range(HEIGHT-1):
        for x in range(WIDTH):
            if snow[y][x]:
                new_y = y + 1
                if new_y < HEIGHT:
                    new_snow[new_y][x] = True
    # случайные новые снежинки сверху
    for x in range(WIDTH):
        if random.random() < 0.02:
            new_snow[0][x] = True
    return new_snow

# отрисовка
def draw(snow, art):
    canvas = [[" "] * WIDTH for _ in range(HEIGHT)]
    # рисуем снег
    for y in range(HEIGHT):
        for x in range(WIDTH):
            if snow[y][x]:
                canvas[y][x] = "*"
    # рисуем 2026 примерно в центре
    start_x = (WIDTH - len(art[0])) // 2
    start_y = 5
    for i, line in enumerate(art):
        for j, c in enumerate(line):
            if c != " " and 0 <= start_y+i < HEIGHT and 0 <= start_x+j < WIDTH:
                canvas[start_y + i][start_x + j] = c
    return "\n".join("".join(row) for row in canvas)

def clear():
    os.system("cls" if os.name == "nt" else "clear")

# main loop
snow = init_snow()
art = get_2026_art()
while True:
    clear()
    print(draw(snow, art))
    snow = update_snow(snow)
    time.sleep(0.1)

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

Анимация по коду выше 👆👆👆

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

🔧 Автоматизация управления репозиториями с GitHub API

from github import Github
from github.GithubException import GithubException
import os

def create_repository(repo_name: str, private: bool = True) -> str:
    try:
        # Получаем токен доступа из переменных окружения
        token = os.getenv("GITHUB_ACCESS_TOKEN")
        if not token:
            raise ValueError("Токен доступа не установлен в переменных окружения.")

        # Подключаемся к GitHub с использованием токена
        g = Github(token)
        user = g.get_user()

        # Создаем новый репозиторий
        repo = user.create_repo(name=repo_name, private=private)
        return f"Репозиторий '{repo.name}' успешно создан по адресу {repo.html_url}"
    except GithubException as e:
        return f"Ошибка при создании репозитория: {e.data['message']}"
    except Exception as e:
        return f"Произошла ошибка: {str(e)}"

# Пример использования
if __name__ == "__main__":
    print(create_repository("my-new-repo", private=True))

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

🛠 pip install PyGithub

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

Если кто есть на reddit, лайкните пост плиз, мне надо набивать там карму 🙏

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

⚡ Быстрая обработка JSON с помощью ujson и orjson

import ujson
import orjson

def load_json_ujson(data: str):
    try:
        return ujson.loads(data)
    except ValueError as e:
        print(f"Ошибка загрузки JSON через ujson: {e}")
        return None

def load_json_orjson(data: bytes):
    try:
        return orjson.loads(data)
    except orjson.JSONDecodeError as e:
        print(f"Ошибка загрузки JSON через orjson: {e}")
        return None

def dump_json_ujson(data: dict) -> str:
    try:
        return ujson.dumps(data)
    except ValueError as e:
        print(f"Ошибка сериализации JSON через ujson: {e}")
        return ""

def dump_json_orjson(data: dict) -> bytes:
    try:
        return orjson.dumps(data)
    except orjson.JSONEncodeError as e:
        print(f"Ошибка сериализации JSON через orjson: {e}")
        return b""

# Пример использования
json_string = '{"name": "Alice", "age": 30}'
json_bytes = b'{"name": "Bob", "age": 25}'

ujson_data = load_json_ujson(json_string)
orjson_data = load_json_orjson(json_bytes)

print(dump_json_ujson(ujson_data))
print(dump_json_orjson(orjson_data))

📌 С помощью библиотек ujson и orjson можно значительно ускорить сериализацию и десериализацию JSON-данных по сравнению со стандартной библиотекой json. Эти библиотеки полезны для повышения производительности веб-приложений и скриптов, работающих с большими объёмами данных. Они обеспечивают более быструю обработку JSON, что критически важно для современного веб-разработчика.

🛠 pip install ujson orjson

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