🔧 Создание простого REST API с Flask: от нуля до деплоя 🚀

from flask import Flask, jsonify, request, abort

app = Flask(__name__)

# Пример данных
tasks = [
    {'id': 1, 'title': 'Сделать стирку', 'done': False},
    {'id': 2, 'title': 'Изучить Flask', 'done': False}
]

@app.route('/api/tasks', methods=['GET'])
def get_tasks():
    return jsonify({'tasks': tasks})

@app.route('/api/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id):
    task = next((task for task in tasks if task['id'] == task_id), None)
    if task is None:
        abort(404)
    return jsonify({'task': task})

@app.route('/api/tasks', methods=['POST'])
def create_task():
    if not request.json or 'title' not in request.json:
        abort(400)
    new_task = {
        'id': tasks[-1]['id'] + 1 if tasks else 1,
        'title': request.json['title'],
        'done': False
    }
    tasks.append(new_task)
    return jsonify({'task': new_task}), 201

@app.route('/api/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
    task = next((task for task in tasks if task['id'] == task_id), None)
    if task is None:
        abort(404)
    if not request.json:
        abort(400)
    task['title'] = request.json.get('title', task['title'])
    task['done'] = request.json.get('done', task['done'])
    return jsonify({'task': task})

@app.route('/api/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    task = next((task for task in tasks if task['id'] == task_id), None)
    if task is None:
        abort(404)
    tasks.remove(task)
    return jsonify({'result': True})

if __name__ == '__main__':
    app.run(debug=True)

📌 Этот код демонстрирует создание простого REST API с Flask, который управляет списком задач. API поддерживает операции CRUD:
получение, добавление, обновление и удаление задач. Такой API может быть полезен для приложений по управлению задачами или в
качестве учебного проекта для изучения разработки веб-сервисов.

🛠 Установка библиотек: pip install Flask

🔒 Совет по безопасности: Рассмотрите добавление аутентификации и авторизации для защиты API. Используйте инструменты, такие
как Postman, для тестирования API.

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

Наши полезные opensource тулы для создания RAG систем и ИИ-агентов:

ChunkTester - подбирает оптимальные чанки под документы и вопросы, и формирует авто-пул тестовых вопросов.

Clusteroid - находит кластера в ваших документах.

Больше по теме в канале @aigentto 🔥

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

🌐 Быстрое создание RESTful API с FastAPI

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

app = FastAPI()

# Модель данных с помощью Pydantic
class Item(BaseModel):
    id: int
    name: str
    description: Optional = None  # Используем Optional для указания, что описание может быть None

# "База данных" в памяти
fake_db = []

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    fake_db.append(item)
    return item

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

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    for item in fake_db:
        if item.id == item_id:
            return item
    raise HTTPException(status_code=404, detail="Item not found")

@app.delete("/items/{item_id}", response_model=Item)
async def delete_item(item_id: int):
    for item in fake_db:
        if item.id == item_id:
            fake_db.remove(item)
            return item
    raise HTTPException(status_code=404, detail="Item not found")

# Пример использования
# Запустите приложение с помощью: uvicorn main:app --reload
# Откройте браузер и перейдите на http://127.0.0.1:8000/docs для интерактивной документации

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

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

🛠 Автоматизация с помощью Python: пакетное переименование файлов

import os

def rename_files_in_directory(directory: str, prefix: str):
    try:
        files = os.listdir(directory)
        for count, filename in enumerate(files):
            old_path = os.path.join(directory, filename)
            new_filename = f"{prefix}_{count}{os.path.splitext(filename)[1]}"
            new_path = os.path.join(directory, new_filename)
            os.rename(old_path, new_path)
        print(f"Успешно переименовано {len(files)} файлов.")
    except FileNotFoundError:
        print("Указанная директория не найдена.")
    except PermissionError:
        print("Отказано в доступе. Проверьте права доступа к директории.")
    except Exception as e:
        print(f"Произошла ошибка: {e}")

# Пример использования
# rename_files_in_directory("/path/to/your/folder", "new_prefix")

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

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

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

from PIL import Image, ImageFilter

def process_image(input_path: str, output_path: str):
    try:
        with Image.open(input_path) as img:
            # Конвертируем изображение в оттенки серого
            gray_img = img.convert('L')
            # Применяем фильтр размытия
            blurred_img = gray_img.filter(ImageFilter.GaussianBlur(2))
            # Сохраняем обработанное изображение
            blurred_img.save(output_path)
        print(f"Изображение сохранено в {output_path}")
    except FileNotFoundError:
        print(f"Файл {input_path} не найден.")
    except PermissionError:
        print("Отказано в доступе. Проверьте права доступа к файлу.")
    except Exception as e:
        print(f"Произошла ошибка: {e}")

# Пример использования
# process_image("path/to/input/image.jpg", "path/to/output/processed_image.jpg")

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

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

Работаем над уровнем 3 и 4

Ранее я писал про будущее Agentic AI. Мы сейчас работаем над уровнями 3 и 4. Скоро будем проводить активное тестирование в полях на двух очень больших компаниях с большими данными и очень сложными workflow. Если подход покажет себя хорошо в этих случаях, то все остальные простые кейсы будут ему тоже по зубам.

Суть здесь в том, чтобы делегировать максимум полномочий ИИ-агентам, иначе зачем мы прикручиваем УМНОГО ИИ-агента и обкладываем его жесткими протоколами и жестким роутингом?! Это то же самое, что нанять кандидата наук и зарегулировать его работу жесткими инструкциями в стиле MacDonalds 🤷‍♂️

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

🛠️ Создаём RESTful TODO API

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

app = FastAPI()

class TodoItem(BaseModel):
    id: int
    title: str
    completed: bool = False

# Хранение данных в памяти для простоты
todo_db: List[TodoItem] = []

@app.post("/todos/", response_model=TodoItem, status_code=201)
async def create_todo(item: TodoItem):
    todo_db.append(item)
    return item

@app.get("/todos/", response_model=List[TodoItem])
async def read_todos():
    return todo_db

@app.get("/todos/{todo_id}", response_model=TodoItem)
async def read_todo(todo_id: int):
    for todo in todo_db:
        if todo.id == todo_id:
            return todo
    raise HTTPException(status_code=404, detail="Задача не найдена")

@app.put("/todos/{todo_id}", response_model=TodoItem)
async def update_todo(todo_id: int, item: TodoItem):
    for idx, todo in enumerate(todo_db):
        if todo.id == todo_id:
            todo_db[idx] = item
            return item
    raise HTTPException(status_code=404, detail="Задача не найдена")

@app.delete("/todos/{todo_id}", status_code=204)
async def delete_todo(todo_id: int):
    for idx, todo in enumerate(todo_db):
        if todo.id == todo_id:
            del todo_db[idx]
            return
    raise HTTPException(status_code=404, detail="Задача не найдена")

# Пример использования: запустите сервер и создайте задачу через POST-запрос

📌 Этот код демонстрирует создание простого RESTful API для управления списком задач (to-do list) с использованием FastAPI. Он включает CRUD-операции: создание, чтение, обновление и удаление задач. FastAPI автоматически обрабатывает валидацию данных и управление ошибками, что упрощает разработку безопасных и эффективных API. Такой API может быть полезен для мобильных приложений, веб-сервисов или интеграции с другими системами.

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

Сходил на закрытую конференцию Yandex.

Вынес оттуда самое ценное — вот этот стикер 😇

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

📊 Генерация отчетов в формате PDF с использованием ReportLab

from reportlab.lib.pagesizes import letter
from reportlab.lib import colors
from reportlab.platypus import SimpleDocTemplate, Table, TableStyle

def create_pdf_report(file_path: str, data: list):
    try:
        document = SimpleDocTemplate(file_path, pagesize=letter)
        elements = []

        table = Table(data)
        table.setStyle(TableStyle([
            ('BACKGROUND', (0, 0), (-1, 0), colors.grey),
            ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke),
            ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
            ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'),
            ('BOTTOMPADDING', (0, 0), (-1, 0), 12),
            ('BACKGROUND', (0, 1), (-1, -1), colors.beige),
            ('GRID', (0, 0), (-1, -1), 1, colors.black)
        ]))

        elements.append(table)
        document.build(elements)
        print(f"PDF report created successfully at {file_path}")

    except Exception as e:
        print(f"An error occurred: {e}")

# Пример использования
data = [
    ["ID", "Name", "Role"],
    [1, "Alice", "Developer"],
    [2, "Bob", "Designer"],
    [3, "Charlie", "Manager"]
]
create_pdf_report("report.pdf", data)

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

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

🧪 Тестирование веб-приложений с Selenium и Pytest

from selenium import webdriver
from selenium.webdriver.common.by import By
import pytest

@pytest.fixture
def browser():
    driver = webdriver.Chrome()
    yield driver
    driver.quit()

def test_google_search(browser):
    browser.get("https://www.google.com")
    search_box = browser.find_element(By.NAME, "q")
    search_box.send_keys("Python")
    search_box.submit()
    
    results = browser.find_elements(By.CSS_SELECTOR, "div.g")
    assert len(results) > 0, "Результаты поиска не найдены."

# Запуск теста: pytest имя_файла.py

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

🛠 Установите библиотеки с помощью команды: pip install selenium pytest (Также потребуется установить ChromeDriver для работы с
браузером Chrome)

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

🔌 Создание RESTful API для управления IoT устройствами с FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict

app = FastAPI()

# Структура данных для устройства
class Device(BaseModel):
    id: int
    name: str
    status: str

# Хранилище устройств
devices: Dict = {}

@app.post("/devices/", response_model=Device)
def create_device(device: Device):
    if device.id in devices:
        raise HTTPException(status_code=400, detail="Device with this ID already exists")
    devices = device
    return device

@app.get("/devices/{device_id}", response_model=Device)
def read_device(device_id: int):
    if device_id not in devices:
        raise HTTPException(status_code=404, detail="Device not found")
    return devices

@app.put("/devices/{device_id}", response_model=Device)
def update_device(device_id: int, device: Device):
    if device_id not in devices:
        raise HTTPException(status_code=404, detail="Device not found")
    devices = device
    return device

@app.delete("/devices/{device_id}", response_model=Device)
def delete_device(device_id: int):
    if device_id not in devices:
        raise HTTPException(status_code=404, detail="Device not found")
    deleted_device = devices.pop(device_id)
    return deleted_device

# Пример использования
# Запустите сервер: uvicorn script_name:app --reload

📌 Этот код предоставляет базовый RESTful API для управления IoT устройствами с использованием FastAPI. Вы можете добавлять,
читать, обновлять и удалять устройства из хранилища, что позволяет легко управлять состоянием ваших IoT устройств. Такое API может
быть полезно для приложений умного дома, где нужно централизованно контролировать различные устройства.

🛠 Установите необходимые библиотеки: pip install fastapi uvicorn

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

🌦️ Асинхронное взаимодействие с 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 🤖