⚡️ Ускоряем Python с Cython

# Пример Cython кода для ускорения вычислений
def sum_of_squares(int n):
    cdef int i
    cdef long total = 0
    for i in range(n):
        total += i * i
    return total

# Сохраните этот код в файле с расширением .pyx и скомпилируйте с помощью setup.py
# Пример setup.py для компиляции

from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize("example.pyx")
)

# Выполните команду: python setup.py build_ext --inplace
# Пример использования
import example

result = example.sum_of_squares(1000000)
print(result)

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

🛠 pip install cython

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

⚡️ Ускорение Python-кода с помощью Cython

# Пример использования Cython для ускорения вычислений
# Сначала создайте файл example.pyx с этим содержимым:

def calculate_sum(n: int) -> int:
    cdef int i, total = 0
    for i in range(n):
        total += i
    return total

# Затем создайте setup.py для компиляции:

from setuptools import setup
from Cython.Build import cythonize

setup(
    ext_modules=cythonize("example.pyx"),
)

# Для компиляции выполните:
# python setup.py build_ext --inplace

# Пример использования скомпилированного модуля:
import example

result = example.calculate_sum(1000000)
print(result)

📌 Cython позволяет компилировать Python-код в C, значительно увеличивая его производительность. В этом примере мы используем Cython для ускорения функции суммирования чисел. Подобный подход пригодится в проектах, где критически важна скорость выполнения, например, в научных вычислениях или обработке больших объемов данных.

🛠 pip install Cython

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

📊 Оптимизация базы данных: ORM vs. прямые SQL-запросы

import sqlite3
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

# Настройка для прямых SQL-запросов
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()
cursor.execute('CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)')

# Настройка для ORM с использованием SQLAlchemy
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

def add_user_sql(name: str):
    try:
        cursor.execute('INSERT INTO users (name) VALUES (?)', (name,))
        conn.commit()
    except sqlite3.Error as e:
        print(f"SQL Error: {e}")

def add_user_orm(name: str):
    try:
        user = User(name=name)
        session.add(user)
        session.commit()
    except Exception as e:
        session.rollback()
        print(f"ORM Error: {e}")

# Пример использования
add_user_sql('Alice')
add_user_orm('Bob')

# Закрытие соединений
cursor.close()
conn.close()
session.close()

📌 В этом примере показано использование прямых SQL-запросов и ORM для добавления записей в базу данных SQLite. Прямые SQL-запросы дают больше контроля и могут быть быстрее, но менее удобны. ORM, как SQLAlchemy, упрощает работу с базой данных, автоматизируя многие процессы, но может добавить некоторую задержку из-за абстракции.

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

Полезный тул от NVIDIA для penetration тестирования вашего ИИ бота

Гоняет всякие промты, чтобы найти дыры (персональные данные и прочее) в вашем AI боте:
https://github.com/NVIDIA/garak/

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

🔐 Замыкания: Управление состоянием без классов

def create_counter(start: int = 0):
    count = start

    def increment():
        nonlocal count
        count += 1
        return count

    return increment

# Пример использования
counter = create_counter()
print(counter())  # Выведет: 1
print(counter())  # Выведет: 2
print(counter())  # Выведет: 3

📌 Замыкания позволяют создавать функции с собственным состоянием, что может быть полезно для управления счетчиками, кэшами и другими задачами, требующими сохранение состояния между вызовами. В приведенном примере замыкание create_counter создает функцию increment, которая помнит текущее значение счетчика. Это позволяет обойтись без использования классов, сохраняя код компактным и понятным.

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

🔒 Автоматизация сетевых операций с Paramiko

import paramiko
from paramiko import SSHException, AuthenticationException, BadHostKeyException

def execute_remote_command(hostname: str, username: str, password: str, command: str) -> str:
    try:
        with paramiko.SSHClient() as client:
            client.set_missing_host_key_policy(paramiko.WarningPolicy())
            client.connect(hostname, username=username, password=password)

            stdin, stdout, stderr = client.exec_command(command)
            result = stdout.read().decode()
            return result
    except AuthenticationException:
        return "Ошибка аутентификации"
    except BadHostKeyException:
        return "Неправильный ключ хоста"
    except SSHException as e:
        return f"SSH ошибка: {e}"
    except Exception as e:
        return f"Ошибка: {e}"

# Пример использования
if __name__ == "__main__":
    response = execute_remote_command(
        hostname='192.168.1.100',
        username='admin',
        password='password',
        command='ls'
    )
    print(response)

📌 Этот скрипт демонстрирует, как выполнять команды на удаленном сервере через SSH с помощью Paramiko. Он автоматически уведомляет о ключах хоста и обрабатывает ошибки подключения. Это полезно для автоматизации задач администрирования и управления серверами.

🛠 pip install paramiko

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

🧠 Быстрый старт с машинным обучением в Scikit-learn

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# Загрузка датасета
data = load_iris()
X, y = data.data, data.target

# Разделение данных на обучающую и тестовую выборки
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Создание и обучение модели
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# Предсказание и оценка модели
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)

print(f"Точность модели: {accuracy:.2f}")

# Обработка ошибок
try:
    # Пример использования модели на новых данных
    new_data = [[5.1, 3.5, 1.4, 0.2]]
    prediction = model.predict(new_data)
    print(f"Предсказанный класс: {prediction[0]}")
except Exception as e:
    print(f"Ошибка при предсказании: {e}")

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

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

🧪 Быстрое тестирование кода с Pytest

# Установка Pytest
# 🛠 pip install pytest

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

# Простой тест для функции add
def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0
    assert add(0, 0) == 0

📌 Pytest — это мощный инструмент для тестирования, который упрощает написание и выполнение тестов. В этом примере показана простая функция сложения и тесты для проверки её корректности. Pytest автоматически обнаруживает и запускает тесты, предоставляя детализированные отчеты об ошибках, что помогает быстро улучшать качество кода.

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

Мониторинг логов с автоматическим исправлением проблем

И так, раньше надо было использовать различные отдельные инструменты для мониторинга логов (Zabbix и прочее), чтобы в крайнем случае сделать reboot отдельных компонентов или всей системы.

Но оказалось, можно отдать задачу ReAct боту типа OpenClaw (мы используем версию B2B DarwinClaw). Он может проверять логи хоть каждые 2 минуты. Тонны токенов здесь не тратятся, он также просто делает поиск по ключевым словам. И важно, что он сам может подобрать ключевые слова и обновлять этот набор исходя из типа логов.

Дальше больше - он может не просто сделать reboot, он может исправить проблему прямо в коде, или сделать откат в git, если поймет, что это проблема новой версии 🔥

Ну то есть, по сути это полноценный support+разраб 24x7, работающий за миску токенов. Да, важно прописать ему некоторые правила (когда то-то делай так, когда это делай сяк), по сути, вашу политику работы с Prod.

#КЕЙС

Прочие полезные кейсы:
👉 WP умер, перевёл все сайты на DarwinClaw
👉 Сбор любой публичной статистики
👉 Отказ от Cursor IDE

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

### Чат-бот для психологической поддержки с использованием NLP 🤖

import random
import spacy

nlp = spacy.load("en_core_web_sm")

def get_response(user_input: str) -> str:
    doc = nlp(user_input)

    # Проверка на наличие ключевых лексем
    for token in doc:
        if token.lemma_ == "sad":
            return random.choice([
                "Мне жаль это слышать. Хочешь поговорить об этом?",
                "Грустно слышать. Может, расскажешь подробнее?",
            ])
        elif token.lemma_ == "happy":
            return random.choice([
                "Это замечательно! Что делает тебя счастливым?",
                "Рад слышать, что ты счастлив!",
            ])
    
    return "Расскажи, как ты себя чувствуешь?"

# Пример использования
def chat():
    print("Чат-бот: Привет! Как ты себя чувствуешь сегодня?")
    while True:
        user_input = input("Вы: ")
        if user_input.lower() in ["выход", "пока"]:
            print("Чат-бот: Пока! Заботься о себе.")
            break
        print(f"Чат-бот: {get_response(user_input)}")

chat()

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

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

🔒 Декораторы для улучшения безопасности API

from functools import wraps

def require_authentication(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        user_authenticated = kwargs.get('authenticated', False)
        if not user_authenticated:
            raise PermissionError("Требуется аутентификация для доступа к этой функции.")
        return func(*args, **kwargs)
    return wrapper

# Пример использования
@require_authentication
def get_user_data(user_id: int, authenticated=False) -> dict:
    # Имитация получения данных пользователя
    return {"user_id": user_id, "name": "Alice"}

try:
    print(get_user_data(user_id=1, authenticated=True))  # Успешный доступ
    print(get_user_data(user_id=2, authenticated=False)) # Ошибка доступа
except PermissionError as e:
    print(e)

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

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

⚡️ Реактивное программирование с RxPy

import rx
from rx import operators as ops
from rx.scheduler import ThreadPoolScheduler
import requests

def fetch_url(url: str) -> str:
    try:
        response = requests.get(url)
        response.raise_for_status()
        return f"Fetched {url} with status {response.status_code}"
    except requests.RequestException as e:
        return f"Error fetching {url}: {str(e)}"

urls = [
    "http://example.com",
    "http://nonexistent.url",
    "http://example.org"
]

# Настройка многопоточного планировщика
scheduler = ThreadPoolScheduler(max_workers=2)

# Создание реактивного потока
rx.from_(urls).pipe(
    ops.map(lambda url: fetch_url(url)),
    ops.do_action(lambda result: print(result))
).subscribe(
    on_error=lambda e: print(f"Error occurred: {e}"),
    on_completed=lambda: print("Fetching completed!")
)

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

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

Два бота OpenClaw

Самим OpenClaw я обычно делаю апдейт его конфигов, но иногда это изменение что-то ломает. Поэтому у меня крутится всегда два бота, один чинит/меняет конфиги другого. А другой первого. Система абсолютно устойчива.

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

🚀 Улучшение производительности веб-приложений с помощью Asyncio

import asyncio
import aiohttp

async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status == 200:
                try:
                    return await response.json()
                except aiohttp.ContentTypeError:
                    print("Получен ответ, не являющийся JSON")
            else:
                response.raise_for_status()

async def main():
    urls = [
        'https://api.example.com/data1',
        'https://api.example.com/data2',
        'https://api.example.com/data3',
    ]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

# Запуск асинхронной программы
asyncio.run(main())

📌 Этот код демонстрирует, как использовать асинхронные операции для выполнения нескольких HTTP-запросов параллельно. С помощью библиотек aiohttp и asyncio можно значительно сократить время отклика при работе с сетевыми запросами, так как они выполняются без блокировки основного потока. Такая техника особенно полезна для веб-приложений, которые часто обращаются к внешним API.

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

🗄️ Оптимизация запросов к базам данных с использованием SQLAlchemy

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship, joinedload

Base = declarative_base()

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    author_id = Column(Integer, ForeignKey('authors.id'))
    author = relationship('Author', back_populates='books')

Author.books = relationship('Book', order_by=Book.id, back_populates='author')

# Настройка базы данных
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# Пример использования: Избегание избыточных запросов с помощью joinedload
def get_books_with_authors():
    with Session() as session:
        try:
            books = session.query(Book).options(joinedload(Book.author)).all()
            return [(book.title, book.author.name) for book in books]
        except Exception as e:
            print(f"Ошибка: {e}")

# Создание примеров данных
with Session() as session:
    author = Author(name='J.K. Rowling')
    book = Book(title='Harry Potter', author=author)
    session.add(author)
    session.add(book)
    session.commit()

# Вызов функции
print(get_books_with_authors())

📌 Этот код демонстрирует использование SQLAlchemy для оптимизации запросов к базе данных с помощью joinedload. Эта техника уменьшает количество SQL-запросов за счет загрузки связанных данных в одно обращение, что особенно полезно при работе с большими объемами данных.

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

🌱 Симуляция клеточного автомата "Жизнь"

import numpy as np
import matplotlib.pyplot as plt
import matplotlib.animation as animation

def update(frame_num, img, grid, N):
    new_grid = grid.copy()
    for i in range(N):
        for j in range(N):
            # Считаем количество живых соседей
            total = (grid[i, (j-1)%N] + grid[i, (j+1)%N] +
                     grid[(i-1)%N, j] + grid[(i+1)%N, j] +
                     grid[(i-1)%N, (j-1)%N] + grid[(i-1)%N, (j+1)%N] +
                     grid[(i+1)%N, (j-1)%N] + grid[(i+1)%N, (j+1)%N])
            # Правила игры
            if grid[i, j] == 1:
                if total < 2 or total > 3:
                    new_grid[i, j] = 0
            elif total == 3:
                new_grid[i, j] = 1
    img.set_data(new_grid)
    grid[:] = new_grid[:]
    return img,

def main():
    N = 100  # Размер сетки
    # Создаем начальную конфигурацию клеток
    grid = np.random.choice([0, 1], N*N, p=[0.8, 0.2]).reshape(N, N)
    fig, ax = plt.subplots()
    img = ax.imshow(grid, interpolation='nearest')
    ani = animation.FuncAnimation(fig, update, fargs=(img, grid, N),
                                  frames=200, interval=300, save_count=50)
    plt.show()

if __name__ == '__main__':
    main()

📌 Этот код демонстрирует реализацию клеточного автомата "Жизнь" Джона Конвея. Алгоритм моделирует эволюцию клеток на двумерной сетке с простыми правилами: клетка остаётся живой, если у неё 2 или 3 живых соседа; иначе умирает. Новый организм рождается, если у пустой клетки ровно 3 живых соседа. Попробуйте изменить размер сетки или вероятность начального состояния, чтобы увидеть разные паттерны.

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

🤖 Создание Telegram чат-бота для автоматизации задач

import logging
from telegram import Update
from telegram.ext import Updater, CommandHandler, CallbackContext

logging.basicConfig(format='%(asctime)s - %(message)s', level=logging.INFO)

def start(update: Update, context: CallbackContext) -> None:
    update.message.reply_text('Привет! Я ваш помощник для автоматизации задач.')

def remind(context: CallbackContext) -> None:
    chat_id = context.job.context
    context.bot.send_message(chat_id=chat_id, text='Время для вашего напоминания!')

def schedule_reminder(update: Update, context: CallbackContext) -> None:
    try:
        delay = int(context.args[0])
        chat_id = update.message.chat_id
        update.message.reply_text(f'Напоминание установлено на {delay} секунд!')
        context.job_queue.run_once(remind, delay, context=chat_id)
    except (IndexError, ValueError):
        update.message.reply_text('Используйте: /remind <время в секундах>')

def main():
    updater = Updater("YOUR_TELEGRAM_BOT_TOKEN", use_context=True)
    dispatcher = updater.dispatcher
    
    dispatcher.add_handler(CommandHandler("start", start))
    dispatcher.add_handler(CommandHandler("remind", schedule_reminder))
    
    updater.start_polling()
    updater.idle()

if __name__ == '__main__':
    main()

📌 Этот код создает Telegram чат-бота, который может устанавливать напоминания. Бот принимает команды от пользователя и позволяет задавать время для напоминания в секундах. Подходит для автоматизации задач, таких как планирование встреч или отправка напоминаний.

🛠 pip install python-telegram-bot

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

🔍 Анализ тональности текстов с помощью NLTK и машинного обучения

import nltk
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
from sklearn.feature_extraction.text import CountVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

nltk.download('punkt')
nltk.download('stopwords')

# Пример данных
texts = ["I love this product!", "This is the worst experience ever.", "Amazing quality, will buy again!", "Not worth the price."]
labels = [1, 0, 1, 0]  # 1 - положительный, 0 - отрицательный

# Предварительная обработка текстов
def preprocess_text(text):
    tokens = word_tokenize(text.lower())
    tokens = [word for word in tokens if word.isalpha()]
    return [word for word in tokens if word not in stopwords.words('english')]

processed_texts = [" ".join(preprocess_text(text)) for text in texts]

# Векторизация текстов и обучение модели
vectorizer = CountVectorizer()
X = vectorizer.fit_transform(processed_texts)
X_train, X_test, y_train, y_test = train_test_split(X, labels, test_size=0.25, random_state=42)

model = MultinomialNB()
model.fit(X_train, y_train)

# Тестирование модели
predictions = model.predict(X_test)
print(f"Точность: {accuracy_score(y_test, predictions)}")

# Применение к новому тексту
new_text = "I am very happy with this purchase!"
processed_new_text = vectorizer.transform([" ".join(preprocess_text(new_text))])
print(f"Тональность: {'Положительная' if model.predict(processed_new_text)[0] == 1 else 'Отрицательная'}")

📌 Этот код использует NLTK для обработки текстов и Naive Bayes для анализа настроения. Это полезно для автоматической оценки отзывов и комментариев, что может помочь в бизнесе для понимания клиентской лояльности.

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

📊 Анализ настроений с NLTK

import nltk
from nltk.sentiment import SentimentIntensityAnalyzer

# Убедитесь, что необходимые ресурсы загружены
nltk.download('vader_lexicon')

def analyze_sentiment(text: str) -> str:
    sia = SentimentIntensityAnalyzer()
    sentiment_scores = sia.polarity_scores(text)
    compound_score = sentiment_scores['compound']
    
    if compound_score >= 0.05:
        return 'Positive'
    elif compound_score <= -0.05:
        return 'Negative'
    else:
        return 'Neutral'

# Пример использования
text = "I absolutely love this new phone!"
print(f"The sentiment of the text is: {analyze_sentiment(text)}")

📌 Этот код использует библиотеку Natural Language Toolkit (NLTK) для анализа настроений в текстах. С помощью SentimentIntensityAnalyzer и лексикона VADER, он оценивает полярность текста, классифицируя его как позитивный, негативный или нейтральный. Это полезно для анализа отзывов клиентов или комментариев в социальных сетях.

Перед использованием кода установите NLTK: pip install nltk и загрузите ресурсы: nltk.download('vader_lexicon').

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

Проблемы внедрения AI в компании

Уже давно есть все инструменты - ИИ, ИИ агенты, мультиагентные системы, и наконец паттерн ReAct (OpenClaw). Последний реально может довольно быстро автоматизировать почти любые процессы.

Но реальное внедрение буксует, вот некоторые причины:
👉 Саботаж и отказ сотрудников что-то менять (автокассы в магазинах тоже прошли этот путь), ведь могут по сути автоматизировать и уволить сотрудников.
👉 Нехватка IT-опыта (когда этим занимаются не IT-специалисты, а вайбкодеры)
👉 Отказ IT-специалистов что-то менять (удивительно, но я сталкиваюсь с жутким консерватизмом со стороны самих IT-специалистов)
👉 Боязнь потратить много токенов (жадность?)

Но что показал эксперимент с roll-up компаниями типа Dwelly, оказывается, даже в небольших бизнесах, которые купил новый собственник, можно все перевести на рельсы AI OS под конкретную вертикаль, и прибыльность бизнеса вырастает с 10% до 40% в год (это в  Великобритании).

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

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

Еще раз - суть в том, чтобы был человек, владеющий долей в компании и имеющий абсолютную власть. Опыт roll-ups показывает, что иногда сопротивляются даже собственники, а вот когда у тебя купили компанию, то все сопротивление заканчивается.

Если вам интересна тема, поставьте 👂, я напишу про реальные кейсы провалов и успехов.

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

🔄 Асинхронная работа с API в функциональном стиле

import asyncio
import aiohttp
from typing import List

# Функция для выполнения запроса к API
async def fetch_data(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        response.raise_for_status()  # Проверка на ошибки HTTP
        return await response.json()  # Возврат JSON-ответа

# Функция для выполнения запросов ко всем URL-адресам
async def fetch_all_data(urls: List[str]) -> List[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url) for url in urls]  # Создание задач для всех URL
        return await asyncio.gather(*tasks, return_exceptions=True)  # Сбор результатов

# Пример использования
async def main():
    urls = [
        "https://jsonplaceholder.typicode.com/posts/1",
        "https://jsonplaceholder.typicode.com/posts/2",
        "https://jsonplaceholder.typicode.com/posts/3",
    ]
    results = await fetch_all_data(urls)
    for result in results:
        print(result)

# Запуск асинхронного кода
asyncio.run(main())

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

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