👍6🤔4🤨3
Самый быстрый способ это узнать — заглянуть в документацию Python. Согласно официальной документации, если подстрока не найдена, метод `find()` возвращает -1.
Теперь интересный момент: что произойдет, если преобразовать `-1` в логическое значение? Давайте проверим:
print(bool(-1)) # Результат: True
Как видите, -1 в Python считается истинным значением (True) при приведении к типу bool.
Please open Telegram to view this post
VIEW IN TELEGRAM
👍7❤4🔥3
from bs4 import BeautifulSoup
html_content = """
<!DOCTYPE html>
<html lang="en">
<head>
<title>Example Page</title>
<meta name="description" content="Example description">
</head>
<body>
<h1>This is a heading</h1>
<p>This is some text in a paragraph.</p>
<a href="https://www.example.com">This is a link</a>
<ul>
<li>Item 1</li>
<li>Item 2</li>
</ul>
<div class="my-class">This is a div</div>
<p id="my-id">This is a paragraph with id </p>
</body>
</html>
"""
soup = BeautifulSoup(html_content, "html.parser") # "html.parser" - используемый парсер
# Навигация и поиск элементов:
title = soup.title.string # Получение текста заголовка
print("Title:", title)
description = soup.find("meta", attrs={"name": "description"})["content"]
print("Description:", description)
links = soup.find_all("a")
for link in links:
print("Link:", link["href"])
paragraphs = soup.find_all("p")
for p in paragraphs:
print("Paragraph:", p.text)
list_items = soup.find("ul").find_all("li")
for item in list_items:
print("List item:", item.text)
div_with_class = soup.find("div", class_="my-class")
print("Div with class:", div_with_class.text)
paragraph_with_id = soup.find("p", id="my-id")
print("Paragraph with id:", paragraph_with_id.text)
# Модификация HTML:
new_paragraph = soup.new_tag("p")
new_paragraph.string = "This is a new paragraph."
soup.body.append(new_paragraph)
print(soup.prettify()) # Вывод измененного HTML с отступами
BeautifulSoup из HTML-строки, получение заголовка, поиск элементов по тегам, атрибутам и классам, извлечение текста и атрибутов элементов, а также добавление нового элемента в HTML-структуру. Используется парсер "html.parser", который встроен в Python. Для более сложных случаев можно использовать другие парсеры, такие как lxml или html5lib. Вывод soup.prettify() позволяет увидеть форматированный HTML-код после внесенных изменений.Please open Telegram to view this post
VIEW IN TELEGRAM
❤6👍6🔥4
* Интеграция с C/C++ библиотеками: вызов функций из библиотек, написанных на C/C++, непосредственно из Python.
* Оптимизация производительности: реализация критичных к производительности участков кода на C/C++ и вызов их из Python.
* Обработка событий: использование функций Python в качестве обработчиков событий в C/C++ приложениях.
* Расширение функциональности Python: добавление новых возможностей в Python с помощью C/C++.
import ctypes
import time
# Загрузка библиотеки C (libcallback_example.so - пример, нужно скомпилировать код C)
lib = ctypes.CDLL("./libcallback_example.so")
# Определение типа функции обратного вызова
CMPFUNC = ctypes.CFUNCTYPE(None, ctypes.c_int)
# Функция Python, которая будет использоваться в качестве обратного вызова
def my_callback(value):
print(f"Python callback called with value: {value}")
# Регистрация функции обратного вызова в библиотеке C
# Предполагается, что в библиотеке C есть функция register_callback, принимающая указатель на функцию
register_callback = lib.register_callback
register_callback.argtypes = [CMPFUNC]
register_callback.restype = None
register_callback(CMPFUNC(my_callback))
# Вызов функции C, которая будет периодически вызывать функцию обратного вызова
# Предполагается, что в библиотеке C есть функция run_loop, которая запускает цикл
run_loop = lib.run_loop
run_loop.argtypes = []
run_loop.restype = None
run_loop()
Пример кода на C (callback_example.c):
#include <stdio.h>
#include <unistd.h> // for sleep
typedef void (*callback_function)(int);
static callback_function registered_callback = NULL;
void register_callback(callback_function callback) {
registered_callback = callback;
}
void run_loop() {
int i = 0;
while (i < 5) {
sleep(1);
if (registered_callback != NULL) {
registered_callback(i);
}
i++;
}
}
Компиляция кода на C:
gcc -shared -o libcallback_example.so -fPIC callback_example.c
ctypes для вызова функций из динамически подключаемой библиотеки C и передачи функции Python в качестве обратного вызова. Код на C регистрирует коллбек и периодически вызывает его. Важно правильно определить типы данных с помощью ctypes.CFUNCTYPE и argtypes/restype. Перед запуском Python кода необходимо скомпилировать код C в shared library (.so файл для Linux).Please open Telegram to view this post
VIEW IN TELEGRAM
❤4👍4🔥3
@dataclass, который автоматически генерирует шаблонный код для классов, преимущественно предназначенных для хранения данных. Это упрощает создание классов с атрибутами, методами __init__, __repr__, __eq__ и другими, без необходимости писать их вручную.from dataclasses import dataclass, field
@dataclass(order=True) # order=True добавляет методы сравнения
class Person:
name: str
age: int
city: str = "Unknown" # Значение по умолчанию
email: str = field(default="no_email@example.com", repr=False) # repr=False скрывает поле в __repr__
scores: list[int] = field(default_factory=list) # mutable default value
person1 = Person("Alice", 30, "New York", scores=[100, 90])
person2 = Person("Bob", 25, "Los Angeles")
person3 = Person("Charlie", 35, scores=[80, 85])
person4 = Person("Alice", 30, "New York", scores=[100, 90])
print(person1) # Вывод: Person(name='Alice', age=30, city='New York', scores=[100, 90])
print(person2) # Вывод: Person(name='Bob', age=25, city='Los Angeles', scores=[])
print(person3) # Вывод: Person(name='Charlie', age=35, city='Unknown', scores=[80, 85])
print(person1 == person4) # True
print(person1 > person2) # True (сравнение по name, затем age)
# Изменение атрибутов:
person1.city = "San Francisco"
person1.scores.append(95)
print(person1) # Вывод: Person(name='Alice', age=30, city='San Francisco', scores=[100, 90, 95])
# Post-init processing
@dataclass
class InventoryItem:
"""Class for keeping track of an item in inventory."""
name: str
unit_price: float
quantity_on_hand: int = 0
def __post_init__(self):
self.total_cost = self.unit_price * self.quantity_on_hand
item = InventoryItem("Pen", 1.5, 50)
print(item.total_cost) # 75.0
Person с различными типами полей, включая значение по умолчанию, field для скрытия поля в __repr__ и default_factory для изменяемых значений по умолчанию. Демонстрируется вывод информации о dataclass с помощью print, сравнение объектов dataclass и изменение атрибутов. Также приведен пример использования __post_init__ для выполнения действий после инициализации dataclass, как, например, вычисление total_cost в InventoryItem. order=True позволяет сравнивать экземпляры dataclass.Please open Telegram to view this post
VIEW IN TELEGRAM
👍8❤4🔥3
from fastapi import FastAPI, Query, Path, HTTPException
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI(title="My Awesome API", version="1.0.0", description="Example FastAPI application")
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None
@app.get("/") # GET-запрос на корневой путь
async def root():
return {"message": "Hello World"}
@app.get("/items/{item_id}") # GET-запрос с параметром пути
async def read_item(item_id: int = Path(..., title="The ID of the item to get", ge=1)): # Валидация параметра пути
return {"item_id": item_id}
@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, min_length=3, max_length=50)): # Валидация query-параметра
results = {"items": [{"item_id": "foo"}, {"item_id": "bar"}]}
if q:
results.update({"q": q})
return results
@app.post("/items/", response_model=Item) # POST-запрос с валидацией тела запроса
async def create_item(item: Item):
return item
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
return {"item_id": item_id, **item.dict()}
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# Пример обработки ошибок
if user_id == 404:
raise HTTPException(status_code=404, detail="User not found")
return {"user_id": user_id}
# Запуск приложения
uvicorn main:app --reload
HTTPException. Для запуска приложения используйте команду uvicorn main:app --reload, где main — имя файла, а app — имя экземпляра FastAPI. После запуска перейдите по адресу /docs в браузере, чтобы увидеть автоматически сгенерированную интерактивную документацию.Please open Telegram to view this post
VIEW IN TELEGRAM
❤8👍4🔥3
difflib разнообразно: от сравнения версий файлов и поиска плагиата до создания патчей и слияния изменений в системах контроля версий. Он также может быть полезен для анализа текстовых данных, например, для выявления изменений в документах или сравнения ответов пользователей.difflib для сравнения двух строк:import difflib
text1 = """
This is the first line.
This is the second line.
This is the third line.
"""
text2 = """
This is the first line.
This is the modified second line.
This is the fourth line.
"""
diff = difflib.ndiff(text1.splitlines(), text2.splitlines())
print("\n".join(diff))
# Вывод:
# This is the first line.
# - This is the second line.
# + This is the modified second line.
# ? +++++++++
#
# - This is the third line.
# ? ---
#
# + This is the fourth line.
# ? ++++
# ^
# Более наглядное сравнение:
diff = difflib.unified_diff(text1.splitlines(), text2.splitlines(), fromfile='file1.txt', tofile='file2.txt')
print("\n".join(diff))
# Вывод:
# --- file1.txt
# +++ file2.txt
# @@ -1,4 +1,4 @@
#
# This is the first line.
# -This is the second line.
# +This is the modified second line.
# -This is the third line.
# +This is the fourth line.
# Сравнение HTML:
from difflib import HtmlDiff
html_diff = HtmlDiff()
table = html_diff.make_table(text1.splitlines(), text2.splitlines(), fromdesc='file1.txt', todesc='file2.txt')
with open('diff.html', 'w') as f:
f.write(table) # Сохранение результата в HTML файл
ndiff и unified_diff для сравнения двух строк построчно. ndiff показывает различия с помощью специальных символов: - для удаленных строк, + для добавленных и ? для изменений внутри строки. unified_diff предоставляет более компактный вывод, похожий на формат патчей. Также показан пример использования HtmlDiff для создания HTML-таблицы с различиями, что удобно для визуального сравнения.Please open Telegram to view this post
VIEW IN TELEGRAM
❤6🔥5👍3
concurrent.futures особенно полезно при работе с I/O-bound задачами, такими как сетевые запросы, чтение/запись файлов, взаимодействие с базами данных. Он также может быть эффективен для CPU-bound задач, если они могут быть разделены на независимые подзадачи.concurrent.futures с ThreadPoolExecutor:import concurrent.futures
import time
import requests
URLS = [
"https://www.example.com",
"https://www.google.com",
"https://www.python.org",
"https://www.wikipedia.org",
]
def load_url(url, timeout):
try:
response = requests.get(url, timeout=timeout)
return f"{url}: {len(response.content)} bytes"
except requests.exceptions.RequestException as e:
return f"{url}: Error: {e}"
start_time = time.time()
# Последовательное выполнение:
for url in URLS:
print(load_url(url, timeout=60))
end_time = time.time()
print(f"Последовательное выполнение заняло: {end_time - start_time} секунд")
start_time = time.time()
# Асинхронное выполнение с ThreadPoolExecutor:
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
# map() возвращает результаты в порядке URLS
future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
except Exception as exc:
print(f"{url}: generated an exception: {exc}")
else:
print(data)
end_time = time.time()
print(f"Асинхронное выполнение заняло: {end_time - start_time} секунд")
ThreadPoolExecutor. Функция load_url загружает данные по указанному URL. ThreadPoolExecutor создает пул потоков, которые выполняют загрузку URL-адресов параллельно. as_completed позволяет обрабатывать результаты по мере их готовности, не дожидаясь завершения всех задач. Сравнение времени выполнения показывает преимущество асинхронного подхода, особенно при работе с сетевыми запросами.Please open Telegram to view this post
VIEW IN TELEGRAM
👍8❤5🔥2
—Декораторный подход, который выглядит элегантно и интуитивно понятен
—Автоматическая генерация справки и сообщений об ошибках
—Вложенные команды из коробки
—Поддержка автодополнения в shell
import click
@click.group()
def cli():
"""Утилита для работы с файлами"""
pass
@cli.command()
@click.argument('path')
@click.option('--lines', '-l', is_flag=True, help='Показать количество строк')
def analyze(path, lines):
"""Анализирует файл и выводит статистику"""
try:
with open(path) as f:
content = f.readlines()
if lines:
click.echo(f'Количество строк: {len(content)}')
except FileNotFoundError:
click.secho('Файл не найден! 😱', fg='red')
if __name__ == '__main__':
cli()
Крутая штука в Click – это то, как легко добавлять новые команды. Хотите добавить подкоманду? Просто навешиваете еще один декоратор!
@cli.command()
@click.argument('path')
def process(path):
"""Обрабатывает файлы с прогресс-баром"""
files = os.listdir(path)
with click.progressbar(files, label='Обработка файлов') as bar:
for f in bar:
# что-то делаем с файлом
time.sleep(0.1)
Click отлично работает с цветным выводом. Хотите привлечь внимание пользователя? Используйте click.secho():
click.secho('Внимание! 🚨', fg='yellow', bold=True)
click.secho('Ошибка! ❌', fg='red')
click.secho('Успех! ✅', fg='green')if click.confirm('Уверены, что хотите удалить все файлы? 🤔'):
click.echo('Поехали! 🚀')Please open Telegram to view this post
VIEW IN TELEGRAM
❤6🔥6👍5🤔1
os и упрощает выполнение распространенных операций, таких как копирование, перемещение, удаление, архивирование и работа с файловыми путями.import shutil
import os
# Копирование файла:
shutil.copy("source_file.txt", "destination_folder/") # Копирование в директорию
shutil.copy("source_file.txt", "destination_file.txt") # Копирование с новым именем
# Копирование директории (рекурсивно):
shutil.copytree("source_folder", "destination_folder")
# Перемещение файла или директории:
shutil.move("source_file.txt", "destination_folder/")
shutil.move("source_folder", "destination_folder")
# Удаление файла:
os.remove("file_to_delete.txt")
# Удаление директории (рекурсивно):
shutil.rmtree("directory_to_delete")
# Создание архива:
shutil.make_archive("archive_name", "zip", "folder_to_archive") # ZIP архив
shutil.make_archive("archive_name", "gztar", "folder_to_archive") # tar.gz архив
shutil.make_archive("archive_name", "bztar", "folder_to_archive") # tar.bz2 архив
shutil.make_archive("archive_name", "xztar", "folder_to_archive") # tar.xz архив
# Работа с файловыми путями (некоторые примеры):
print(shutil.which("python")) # Путь к исполняемому файлу python
# Проверка дискового пространства
total, used, free = shutil.disk_usage("/") # Получение информации о дисковом пространстве (корневой раздел)
print(f"Total: {total // (2**30)} GiB")
print(f"Used: {used // (2**30)} GiB")
print(f"Free: {free // (2**30)} GiB")
shutil для копирования файлов и директорий, перемещения, удаления, создания архивов различных форматов, а также примеры работы с файловыми путями и проверки дискового пространства. Обратите внимание, что операции с файлами и директориями могут привести к безвозвратной потере данных, поэтому будьте осторожны при их использовании. Перед выполнением операций на реальных данных рекомендуется протестировать код на тестовых файлах и директориях. Также показан пример использования shutil.which для определения пути к исполняемому файлу и shutil.disk_usage для получения информации о дисковом пространстве.Please open Telegram to view this post
VIEW IN TELEGRAM
❤5👍5🔥4
import numpy as np
# Создание массивов:
a = np.array([1, 2, 3]) # Одномерный массив
b = np.array([[1, 2, 3], [4, 5, 6]]) # Двумерный массив
c = np.zeros((2, 3)) # Массив из нулей
d = np.ones((3, 2)) # Массив из единиц
e = np.arange(10) # Массив с числами от 0 до 9
f = np.linspace(0, 1, 5) # Массив из 5 чисел, равномерно распределенных от 0 до 1
g = np.random.rand(2, 2) # Массив случайных чисел от 0 до 1
print("a:", a)
print("b:", b)
print("c:", c)
print("d:", d)
print("e:", e)
print("f:", f)
print("g:", g)
# Математические операции:
print("a + 2:", a + 2)
print("a * 3:", a * 3)
print("a.sum():", a.sum()) # Сумма элементов массива
print("b.mean():", b.mean()) # Среднее значение элементов массива
print("a.dot(a): ", a.dot(a)) # Скалярное произведение векторов
# Индексация и срезы:
print("a[0]:", a[0])
print("b[1, 2]:", b[1, 2])
print("e[1:4]:", e[1:4])
# Изменение формы массива:
print("b.reshape(3, 2):", b.reshape(3, 2))
print("a.T:", a.T) # Транспонирование массива
# Линейная алгебра:
A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])
print("A.dot(B):", A.dot(B)) # Умножение матриц
print("np.linalg.inv(A):", np.linalg.inv(A)) # Обратная матрица
Please open Telegram to view this post
VIEW IN TELEGRAM
1👍7❤5🔥5
import time
def compute_intensive_task(n):
result = 0
for i in range(n):
result += i * i
return result
start_time = time.time()
result = compute_intensive_task(10000000) # CPU-bound задача
end_time = time.time()
print(f"Результат: {result}")
print(f"Время выполнения: {end_time - start_time} секунд")
# Для запуска с pypy3: pypy3 example.py
Please open Telegram to view this post
VIEW IN TELEGRAM
👍6❤4🔥3
def test_sum():
assert sum([1, 2, 3]) == 6
assert sum([-1, 1]) == 0
—Понятны даже джуниору
—Легко дебажить
—Отлично подходят для документирования кода
—Быстро выполняются
—Покрывают только те кейсы, о которых мы подумали
—Часто пропускаем edge-cases
—Может быть много копипасты
from hypothesis import given
import hypothesis.strategies as st
@given(st.lists(st.integers()))
def test_sum_properties(numbers):
assert sum(numbers) == sum(reversed(numbers))
assert sum(numbers + [0]) == sum(numbers)
—Находит неочевидные баги
—Меньше кода, больше покрытие
—Заставляет думать о свойствах функций, а не о конкретных значениях
—Сложнее придумывать правильные свойства
—Медленнее выполняются
—Может быть сложно понять, почему тест упал
Please open Telegram to view this post
VIEW IN TELEGRAM
👍8❤4🔥3
# Создать новое окружение с Python 3.9
conda create -n my_env python=3.9
# Активировать окружение
conda activate my_env
# Установить пакеты в активное окружение
conda install numpy pandas scipy
# Просмотреть список установленных пакетов
conda list
# Деактивировать окружение
conda deactivate
# Удалить окружение
conda remove -n my_env --all
# Создать окружение из файла environment.yml
conda env create -f environment.yml
# Экспортировать окружение в файл environment.yml
conda env export > environment.yml
# Обновить все пакеты в активном окружении
conda update --all
# Поиск пакета
conda search numpy
environment.yml, экспортировать текущее окружение в файл environment.yml , обновлять все пакеты в окружении и искать нужный пакет. Использование виртуальных окружений с conda — лучшая практика для управления зависимостями проектов и предотвращения конфликтов.Please open Telegram to view this post
VIEW IN TELEGRAM
1👍9❤5🔥3
Path.from pathlib import Path
# Создание объекта Path
file_path = Path("my_file.txt")
directory_path = Path("my_directory")
# Проверка существования файла/директории
if file_path.exists():
print(f"Файл {file_path} существует")
else:
print(f"Файл {file_path} не существует")
if directory_path.exists():
print(f"Директория {directory_path} существует")
if directory_path.is_dir():
print(f"Это директория")
# Создание директории
directory_path.mkdir(parents=True, exist_ok=True) # parents=True создает все родительские директории при необходимости, exist_ok=True предотвращает ошибку, если директория уже существует
# Создание файла и запись в него
file_path.write_text("Hello, pathlib!")
# Чтение из файла
file_content = file_path.read_text()
print(f"Содержимое файла: {file_content}")
# Переименование файла
new_file_path = Path("my_new_file.txt")
file_path.rename(new_file_path)
# Удаление файла
new_file_path.unlink()
# Перебор файлов в директории
for file in directory_path.iterdir():
print(file)
# Получение абсолютного пути
absolute_path = directory_path.resolve()
print(f"Абсолютный путь: {absolute_path}")
# Создание вложенных директорий и файла
nested_dir = Path("my_directory/nested/deeper")
nested_dir.mkdir(parents=True, exist_ok=True)
nested_file = nested_dir / "nested_file.txt" # / используется для объединения путей
nested_file.write_text("Content in nested file")
import shutil
# Удаление директории и ее содержимого
shutil.rmtree(directory_path) # rmtree из shutil используется для удаления непустых директорий
Path, проверяется существование файлов и директорий, создаются и удаляются файлы и директории, производится чтение и запись данных, переименование файлов, перебор файлов в директории и получение абсолютного пути. Также показан пример создания вложенных директорий и файла внутри них с помощью оператора /. И наконец, показано, как удалить директорию и ее содержимое с помощью shutil.rmtree.Please open Telegram to view this post
VIEW IN TELEGRAM
🔥9❤4👍4
Forwarded from IT memer
Please open Telegram to view this post
VIEW IN TELEGRAM
👍10🤣6❤2😁2
httpx. В Python httpx предоставляет мощный и удобный асинхронный интерфейс для работы с HTTP-запросами. Асинхронный режим позволяет эффективно обрабатывать множество запросов параллельно, не блокируя основной поток выполнения, что особенно полезно при взаимодействии с медленными или перегруженными серверами.import asyncio
import httpx
async def fetch_page(client: httpx.AsyncClient, url: str):
try:
response = await client.get(url, timeout=10.0)
response.raise_for_status()
return {"url": url, "status_code": response.status_code, "content": response.text}
except httpx.HTTPError as exc:
return {"url": url, "error": f"HTTP Error: {exc}"}
except httpx.TimeoutException:
return {"url": url, "error": "Request timed out"}
except Exception as e:
return {"url": url, "error": f"An unexpected error occurred: {e}"}
async def main():
urls = [
"https://www.example.com",
"https://www.google.com",
"https://www.python.org",
"https://this-is-an-invalid-url.com", # Пример некорректного URL
"https://jsonplaceholder.typicode.com/todos/1" # Пример API
]
async with httpx.AsyncClient(http2=True) as client: # Использование HTTP/2
tasks = [fetch_page(client, url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
async def make_requests_with_limits(urls, max_concurrent=5):
semaphore = asyncio.Semaphore(max_concurrent)
async def fetch_with_semaphore(url):
async with semaphore:
return await fetch_page(httpx.AsyncClient(), url)
async with httpx.AsyncClient() as client:
tasks = [fetch_with_semaphore(url) for url in urls]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
if __name__ == "__main__":
asyncio.run(main())
# Пример с ограничением числа одновременных запросов
asyncio.run(make_requests_with_limits(["https://www.example.com"] * 10))
fetch_page теперь принимает httpx.AsyncClient в качестве аргумента, что позволяет использовать один клиент для всех запросов внутри main. Это способствует более эффективному использованию соединений. Добавлен пример с некорректным URL и API-запросом для демонстрации обработки ошибок и различных типов ответов. Также показано, как использовать HTTP/2 с помощью httpx.AsyncClient(http2=True). Дополнительно, представлен пример make_requests_with_limits с использованием asyncio.Semaphore для ограничения количества одновременных запросов, что позволяет контролировать нагрузку на сервер и предотвращать ошибки, связанные с превышением лимитов.Please open Telegram to view this post
VIEW IN TELEGRAM
❤6👍5🔥3
from tqdm import tqdm
import time
# Простой пример с range
for i in tqdm(range(100)):
time.sleep(0.02) # Имитация длительной операции
# С ручным обновлением
with tqdm(total=100) as pbar:
for i in range(10):
time.sleep(0.1)
pbar.update(10) # Обновление индикатора на 10 единиц
# С описанием
for i in tqdm(range(100), desc="Обработка данных"):
time.sleep(0.02)
# Вложенные циклы
from tqdm import trange
for i in trange(4, desc="1st loop"):
for j in trange(5, desc="2nd loop", leave=False):
for k in trange(50, desc="3rd loop", leave=False):
time.sleep(0.001)
# С pandas
import pandas as pd
df = pd.DataFrame({'a': range(10000)})
# Применение tqdm к pandas DataFrame (apply с progress_apply)
tqdm.pandas() # Необходимо для progress_apply
df['b'] = df['a'].progress_apply(lambda x: x*x) # Индикатор выполнения для apply
# Пример с файлами
import os
with open("large_file.txt", "w") as f:
for i in trange(10000, desc="Создание файла"):
f.write("This is a line of text.\n")
with open("large_file.txt", "r") as f:
for line in tqdm(f, total=os.path.getsize("large_file.txt"), unit="B", unit_scale=True, unit_divisor=1024, desc="Чтение файла"):
# Обработка каждой строки файла
pass
range, ручным обновлением, описанием, вложенными циклами, интеграцией с pandas DataFrame с помощью progress_apply и обработкой файлов. Обратите внимание на использование tqdm.pandas() для работы с pandas. Пример с файлами показывает, как отображать прогресс чтения/записи с учетом размера файла.Please open Telegram to view this post
VIEW IN TELEGRAM
👍8❤3🔥2
import asyncio
import aioredis
async def redis_example():
try:
# Подключение к Redis
redis = await aioredis.create_redis_pool(
'redis://localhost',
db=0,
encoding='utf-8', # Декодирование в UTF-8
minsize=5, # Минимальное количество соединений в пуле
maxsize=10 # Максимальное количество соединений в пуле
)
# Работа со строками
await redis.set('name', 'John Doe')
name = await redis.get('name')
print(f"Name: {name}") # Вывод: Name: John Doe
# Работа со списками
await redis.rpush('my_list', 'one', 'two', 'three')
list_length = await redis.llen('my_list')
list_items = await redis.lrange('my_list', 0, -1)
print(f"List length: {list_length}") # Вывод: List length: 3
print(f"List items: {list_items}") # Вывод: List items: ['one', 'two', 'three']
# Работа с хэшами
await redis.hset('user:1', 'name', 'Alice')
await redis.hset('user:1', 'age', 30)
user_data = await redis.hgetall('user:1')
print(f"User data: {user_data}") # Вывод: User data: {'name': 'Alice', 'age': '30'}
# Работа с множествами
await redis.sadd('my_set', 'apple', 'banana', 'cherry')
set_members = await redis.smembers('my_set')
print(f"Set members: {set_members}") # Вывод: Set members: {'apple', 'banana', 'cherry'}
# Использование pipeline
async with redis.pipeline(transaction=True) as pipe: # transaction=True для атомарности операций
await pipe.set('key1', 'value1')
await pipe.get('key1')
await pipe.hset('user:2', 'name', 'Bob')
results = await pipe.execute()
print(f"Pipeline results: {results}")
redis.close()
await redis.wait_closed()
except aioredis.RedisError as e:
print(f"Ошибка Redis: {e}")
if __name__ == "__main__":
asyncio.run(redis_example())
aioredis.create_redis_pool, работа с различными типами данных (строки, списки, хэши, множества), и использование pipeline для пакетной обработки команд в транзакции. encoding='utf-8' в параметрах подключения обеспечивает декодирование в UTF-8. minsize и maxsize управляют размером пула соединений. Обработка исключения aioredis.RedisError помогает выявить проблемы с Redis. Важно запускать асинхронный код внутри asyncio.run.Please open Telegram to view this post
VIEW IN TELEGRAM
1❤5👍4🔥1🥰1