Представьте, что ваш код – это оркестр в Мариинском театре. Каждый инструмент (поток или корутина) играет свою партию. А вы – дирижёр, который должен убедиться, что все звучит гармонично. Вот только как это сделать, когда все играют одновременно?
➡️ 1. Изоляция – ключ к успеху
Первое правило тестирования параллельного кода: изолируйте тесты! Каждый тест должен быть как отдельная комната в звукоизолированной студии. Используйте моки и стабы, чтобы симулировать внешние зависимости. Вот пример с использованием unittest.mock:
⬆️ Мы изолировали тест от реального API. Теперь он быстрый, как Усэйн Болт, и предсказуемый, как восход солнца!
➡️ 2. Детерминизм – ваш лучший друг
Асинхронный код может быть непредсказуемым, как погода в Питере. Но ваши тесты должны быть стабильными, как гранитная набережная. Используйте семафоры, события и другие примитивы синхронизации, чтобы контролировать порядок выполнения. Вот пример с использованием asyncio.Event:
⬆️ Этот тест всегда будет проходить, даже если вы запустите его на компьютере, работающем на картофельной батарейке!
➡️ 3. Таймауты – не просто для пиццы
Установка таймаутов в тестах – это как страховка. Вы надеетесь, что она не понадобится, но лучше иметь ее под рукой. Вот как можно использовать таймауты в pytest:
⬆️ Этот тест убедится, что ваша функция не зависнет, как старый Windows при запуске Crysis!
➡️ 4. Асинхронные фикстуры – ваш секретный козырь
В мире async/await фикстуры тоже должны быть асинхронными. Используйте async fixtures в pytest для подготовки и очистки тестового окружения. Вот пример:
⬆️ Эта фикстура – как заботливая мама, которая готовит завтрак перед школой и убирает посуду после. Только вместо завтрака у нас база данных!
➡️ 5. Параллельное выполнение тестов – двойная выгода
Запуск тестов параллельно не только ускоряет процесс, но и помогает выявить проблемы с состоянием гонки. Используйте pytest-xdist, но будьте осторожны: убедитесь, что ваши тесты действительно независимы друг от друга. Вот команда для запуска:
⬆️ Это как устроить гонки Формулы-1 для ваших тестов. Победит самый быстрый и надежный код!
🐍 Pythoner
Первое правило тестирования параллельного кода: изолируйте тесты! Каждый тест должен быть как отдельная комната в звукоизолированной студии. Используйте моки и стабы, чтобы симулировать внешние зависимости. Вот пример с использованием unittest.mock:
from unittest.mock import patch
import asyncio
async def fetch_data(url):
# Реальный запрос к API
...
@patch('your_module.fetch_data')
async def test_process_data(mock_fetch):
mock_fetch.return_value = {'key': 'value'}
result = await process_data('http://api.example.com')
assert result == 'processed value'
Асинхронный код может быть непредсказуемым, как погода в Питере. Но ваши тесты должны быть стабильными, как гранитная набережная. Используйте семафоры, события и другие примитивы синхронизации, чтобы контролировать порядок выполнения. Вот пример с использованием asyncio.Event:
import asyncio
async def test_order_of_execution():
event = asyncio.Event()
results = []
async def task1():
await event.wait()
results.append(1)
async def task2():
results.append(2)
event.set()
await asyncio.gather(task1(), task2())
assert results == [2, 1]
Установка таймаутов в тестах – это как страховка. Вы надеетесь, что она не понадобится, но лучше иметь ее под рукой. Вот как можно использовать таймауты в pytest:
import pytest
import asyncio
@pytest.mark.asyncio
async def test_long_running_task():
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(never_ending_task(), timeout=1.0)
В мире async/await фикстуры тоже должны быть асинхронными. Используйте async fixtures в pytest для подготовки и очистки тестового окружения. Вот пример:
import pytest
import asyncio
@pytest.fixture
async def database():
db = await create_database_connection()
yield db
await db.close()
@pytest.mark.asyncio
async def test_database_query(database):
result = await database.fetch('SELECT * FROM users')
assert len(result) > 0
Запуск тестов параллельно не только ускоряет процесс, но и помогает выявить проблемы с состоянием гонки. Используйте pytest-xdist, но будьте осторожны: убедитесь, что ваши тесты действительно независимы друг от друга. Вот команда для запуска:
pytest -n auto your_test_file.py
Please open Telegram to view this post
VIEW IN TELEGRAM
👍4❤3🔥3
# Сборка образа
docker build -t my-image .
# Запуск контейнера
docker run -d --name my-container my-image
# Просмотр запущенных контейнеров
docker ps
# Остановка контейнера
docker stop my-container
Docker Compose позволяет определять и запускать многоконтейнерные приложения. Пример docker-compose.yml файла:
version: '3'
services:
web:
build: .
ports:
- "5000:5000"
redis:
image: "redis:alpine"
Docker предоставляет мощные возможности для создания сетей между контейнерами:
Volumes используются для хранения данных вне контейнеров:
# Создание volume
docker volume create my-vol
# Использование volume при запуске контейнера
docker run -v my-vol:/app/data my-image
Docker Swarm - это инструмент для оркестрации контейнеров, позволяющий управлять кластером Docker-хостов:
Docker предоставляет мощный инструментарий для контейнеризации приложений, от простых сценариев до сложных микросервисных архитектур. Освоение Docker открывает новые возможности для разработки, тестирования и развертывания программного обеспечения.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6👍5🔥2🤣1
import jmespath
jmespath.search('foo.bar', {'foo': {'bar': 'baz'}})
# output: 'baz'
jmespath.search('foo.*.name', {'foo': {'bar': {'name': 'one'}, 'baz':
{'name': 'two'}}})
# output: ['one', 'two']
Please open Telegram to view this post
VIEW IN TELEGRAM
👍6❤5🔥4
import ast
code = """
def calculate_sum(a, b):
result = a + b
print(f"Sum: {result}")
return result
"""
# Создаем AST
tree = ast.parse(code)
# Анализируем структуру
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
print(f"Найдена функция: {node.name}")
elif isinstance(node, ast.Name):
print(f"Найдена переменная: {node.id}")
class DebugTransformer(ast.NodeTransformer):
def visit_FunctionDef(self, node):
# Добавляем отладочный print в начало каждой функции
debug_print = ast.Expr(
value=ast.Call(
func=ast.Name(id='print', ctx=ast.Load()),
args=[ast.Constant(value=f"Вызов функции {node.name}")],
keywords=[]
)
)
node.body.insert(0, debug_print)
return node
# Применяем трансформацию
transformed = DebugTransformer().visit(tree)
class StringOptimizer(ast.NodeTransformer):
def visit_BinOp(self, node):
# Оптимизация конкатенации строк
if isinstance(node.op, ast.Add):
if isinstance(node.left, ast.Constant) and isinstance(node.right, ast.Constant):
if isinstance(node.left.value, str) and isinstance(node.right.value, str):
return ast.Constant(value=node.left.value + node.right.value)
return node
- Статический анализ безопасности: поиск потенциальных уязвимостей в коде
- Автоматическая документация: генерация документации на основе структуры кода
- Миграция кода: автоматическое обновление устаревших конструкций
- Оптимизация производительности: автоматический поиск неэффективных паттернов
class ComplexityAnalyzer(ast.NodeVisitor):
def __init__(self):
self.complexity = 0
def visit_If(self, node):
self.complexity += 1
self.generic_visit(node)
def visit_For(self, node):
self.complexity += 2
self.generic_visit(node)
def visit_While(self, node):
self.complexity += 2
self.generic_visit(node)
# Использование
analyzer = ComplexityAnalyzer()
analyzer.visit(tree)
print(f"Сложность кода: {analyzer.complexity}")
- Работа с типами данных и аннотациями через AST
- Создание собственных декораторов с помощью трансформации AST
- Оптимизация циклов и условных конструкций
- Анализ потока данных в программе
Please open Telegram to view this post
VIEW IN TELEGRAM
👍7❤3🔥3
- Снижение затрат: Платите только за фактическое использование ресурсов, что часто приводит к экономии средств.
- Масштабируемость: Автоматическое масштабирование помогает справляться с любыми нагрузками без необходимости в ручном управлении.
- Упрощение DevOps: Нет необходимости управлять серверами и их конфигурацией.
- Холодный старт: Время, необходимое для инициализации функции, может быть значительным и влиять на производительность.
- Ограничения исполнения: Функции могут иметь ограничения по времени выполнения и памяти, что может быть проблематично для более сложных задач.
- Зависимость от провайдера: Использование специфичных для провайдера технологий может привести к привязке к одному облаку.
Serverless архитектура предоставляет мощный инструмент для современных разработчиков, предлагая значительную экономию времени и ресурсов. Однако, важно учитывать её ограничения при выборе данной модели для своего проекта.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤4🔥4👍3
—Нет необходимости привязываться к конкретным классам и типам.
—Код становится более гибким и менее связанным.
—Легче расширять и изменять код, добавляя новые типы.
—Упрощает полиморфизм.
class Bird:
def fly(self):
return "I can fly!"
class Duck(Bird):
def quack(self):
return "Quack!"
class Airplane:
def fly(self):
return "I can also fly!"
def make_it_fly(flyable_thing):
print(flyable_thing.fly())
# Создаем объекты
duck = Duck()
airplane = Airplane()
# Используем их, не смотря на разные типы
make_it_fly(duck) # Выведет: I can fly!
make_it_fly(airplane) # Выведет: I can also fly!
make_it_fly принимает любой объект, который имеет метод fly(). Если объект соответствует этому интерфейсу, он будет выполнен, независимо от того, является ли объект уткой, самолетом или чем-то еще.Please open Telegram to view this post
VIEW IN TELEGRAM
👍9❤4🔥3🤣1
—Автоматическое извлечение метрик: Библиотека собирает метрики, такие как распределения, статистики и т.д.
—Поддержка различных форматов данных: Логирование можно выполнять для различных источников данных, включая Pandas DataFrames и потоковые данные.
—Интеграция с другими инструментами: Whylogs легко интегрируется с другими библиотеками и фреймворками.
import whylogs as why
import pandas as pd
# Создание примера данных
data = {
"age": [25, 30, 35, 40, 45],
"income": [50000, 60000, 70000, 80000, 90000]
}
df = pd.DataFrame(data)
# Инициализация логгера
logger = why.logger()
# Логируем данные
logger.log_dataframe(df)
# Генерируем отчет
report = logger.report()
# Сохраняем отчет
report.save("report.whylog")
# Выводим на экран
print(report)
age и income. Затем мы инициализируем Whylogs логгер, логируем наш DataFrame, генерируем и сохраняем отчет.Please open Telegram to view this post
VIEW IN TELEGRAM
❤4👍4🔥2
—Возбуждение существующих исключений: Вы можете использовать
raise без каких-либо параметров в блоке except, чтобы повторно вызвать текущее исключение. —Создание собственных исключений: Вы можете создать или вызвать собственное исключение, создав объект исключения и передав его в
raise.def divide(a, b):
if b == 0:
raise ValueError("Деление на ноль невозможно.")
return a / b
try:
result = divide(10, 0)
except ValueError as e:
print(f"Ошибка: {e}")
divide вызывает исключение ValueError, если второй аргумент равен нулю. В блоке try мы пытаемся выполнить деление, и если возникает ошибка, мы перехватываем её и выводим сообщение об ошибке.Please open Telegram to view this post
VIEW IN TELEGRAM
👍8❤3🔥3
# Без slots
class User:
def __init__(self, name, age):
self.name = name
self.age = age
# Со slots
class OptimizedUser:
__slots__ = ['name', 'age']
def __init__(self, name, age):
self.name = name
self.age = age
—В больших датасетах, где у вас тысячи/миллионы однотипных объектов
—В микросервисах, где важна оптимизация памяти
—В системах реального времени, где каждый байт на счету
—В долгоживущих процессах, обрабатывающих потоки данных
—Нельзя добавлять новые атрибуты после определения класса
—Наследование работает не так прозрачно, как с обычными классами
—Некоторые метаклассы могут конфликтовать со slots
class DataPoint:
__slots__ = ['timestamp', 'value', 'sensor_id']
def __init__(self, timestamp, value, sensor_id):
self.timestamp = timestamp
self.value = value
self.sensor_id = sensor_id
# Представьте, что таких объектов у вас миллионы
data_points = [DataPoint(time.time(), random.random(), i) for i in range(1_000_000)]
Please open Telegram to view this post
VIEW IN TELEGRAM
👍6🔥5❤4
def main():
# Точка входа в программу
if __name__ == '__main__':
main()
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6🔥4👍3
Во время разработки программы, вы можете встретить ситуации, когда вам нужно оставить пустое место для будущего кода. Вместо того чтобы оставлять комментарии, вы можете использовать оператор
pass, чтобы указать на то, что здесь будет код. Это позволяет сохранить структуру программы и избежать ошибок синтаксиса.Оператор
pass может быть полезен при создании минимальных классов или функций. В Python, класс или функция не может быть пустой. Если вы попытаетесь создать пустой класс или функцию, интерпретатор вернет ошибку. Оператор pass позволяет обойти это ограничение.Оператор
pass также может быть использован для управления потоком программы. Иногда в условной конструкции if/elif/else или в цикле for/while может не быть необходимости выполнять какое-либо действие. В этих случаях можно использовать оператор pass для обозначения пустого блока.Please open Telegram to view this post
VIEW IN TELEGRAM
👍10❤4🔥2
def create_power_function(power):
def power_func(x):
return x ** power
return power_func
# Создаём функции на лету
square = create_power_function(2)
cube = create_power_function(3)
import types
def create_dynamic_function(func_name, args, body):
namespace = {}
func_code = f"def {func_name}({', '.join(args)}):\n{body}"
exec(func_code, globals(), namespace)
return namespace[func_name]
# Магия в действии
greet = create_dynamic_function(
"greet",
["name"],
" return f'Привет, {name}!'"
)
—При создании API с повторяющимися паттернами
—Для автоматизации рутинных задач
—При разработке фреймворков и библиотек
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥8❤5👍5
—Объектно-ориентированный подход вместо строковых операций
—Кроссплатформенность из коробки
—Цепочки методов, которые читаются как поэзия
—Меньше кода, больше смысла
# Старый подход с os.path
import os.path
file_path = os.path.join('data', 'users', 'config.json')
parent_dir = os.path.dirname(file_path)
file_name = os.path.basename(file_path)
# Новый подход с pathlib
from pathlib import Path
file_path = Path('data') / 'users' / 'config.json'
parent_dir = file_path.parent
file_name = file_path.name
path = Path('config.json')
if path.exists():
print('Файл существует!')Path('nested/directories/structure').mkdir(parents=True, exist_ok=True)# Найти все .py файлы в текущей директории
python_files = list(Path('.').glob('*.py'))
path = Path('document.pdf')
print(path.suffix) # .pdf
print(path.stem) # documentconfig_path = (Path.home() / 'projects' / 'app' / 'config.json')
if config_path.exists():
data = json.loads(config_path.read_text())
with Path('log.txt').open('w') as f:
f.write('Logging started')Please open Telegram to view this post
VIEW IN TELEGRAM
👍11❤3🔥2
a = 3
b = 10
print('a больше b') if a > b else print('a меньше b')
result = 'Четное' if a % 2 == 0 else 'Нечетное'
print(result)
# Результат:
# a меньше b
# Нечетное
—Для вывода одной из двух фраз в зависимости от условия
a > b.—Для присваивания переменной
result одного из двух значений в зависимости от четности a.Please open Telegram to view this post
VIEW IN TELEGRAM
👍15🔥6❤4
—Веб-приложения для отображения статистики.
—Информационные табло и панели мониторинга.
—Системы аналитики для визуализации данных.
—CLI-интерфейсы для вывода данных пользователю.
import humanize
import datetime
# Пример: время, прошедшее с 24 сентября 2023 года
dt = datetime.datetime(2023, 9, 24)
print(humanize.naturaltime(dt)) # Вывод: "1 месяц назад" (на момент текущей даты)
import humanize
number = 1500
print(humanize.intcomma(number)) # Вывод: "1,500"
print(humanize.intword(number)) # Вывод: "1.5 k"
import humanize
duration = 3661 # время в секундах
print(humanize.precisedelta(datetime.timedelta(seconds=duration)))
# Вывод: "1 час, 1 минутa, 1 секунда"
Please open Telegram to view this post
VIEW IN TELEGRAM
👍10❤4🔥4
os.rename(old_name, new_name)
import os
# Переименуем файл 'старый_файл.txt' в 'новый_файл.txt'
old_name = 'старый_файл.txt'
new_name = 'новый_файл.txt'
# Проверяем, существует ли старый файл
if os.path.exists(old_name):
os.rename(old_name, new_name)
print(f'Файл переименован в {new_name}')
else:
print(f'Файл {old_name} не найден')
—Если файл с
new_name уже существует, функция вызовет ошибку. —Убедитесь, что у вас есть необходимые права для выполнения этой операции.
Please open Telegram to view this post
VIEW IN TELEGRAM
👍13❤5🔥3⚡1
—Узнать размер файла, сравнив позицию после чтения с начальной.
—Вернуться к определенной позиции с помощью
seek().—Отслеживать прогресс обработки больших файлов.
—Получить позицию в строковых буферах и потоках байтов.
# Открываем файл для записи
with open('example.txt', 'w') as f:
f.write('Hello, World!')
# Получаем текущую позицию курсора
position = f.tell()
print(f'Current position after writing: {position}') # Вывод: 13, т.к. длина строки "Hello, World!" = 13
# Открываем файл для чтения
with open('example.txt', 'r') as f:
content = f.read(5) # Читаем первые 5 символов
position = f.tell()
print(f'Content read: {content}') # Вывод: Hello
print(f'Current position after reading: {position}') # Вывод: 5
1. Мы открываем файл
example.txt для записи и записываем в него строку "Hello, World!".2. Используем
tell(), чтобы получить позицию курсора после записи, что будет равно 13 (длина строки).3. Затем открываем файл для чтения и читаем первые 5 символов.
4. Снова используем
tell(), чтобы получить новую позицию курсора, которая будет равна 5 после чтения. Please open Telegram to view this post
VIEW IN TELEGRAM
🔥6👍3❤2
class MyClass:
def __init__(self, name):
self.name = name
print(f"Объект {self.name} создан.")
def __del__(self):
print(f"Объект {self.name} уничтожен.")
# Создаем объект класса
obj = MyClass("TestObject")
# Удаляем объект
del obj
# Если вы хотите избежать автоматического освобождения памяти,
# можете использовать осмысленное завершение программы, например:
import time
time.sleep(1) # daем немного времени, чтобы увидеть сообщение деструктора
MyClass, после чего его деструктор будет вызван при удалении объекта с помощью команды del obj. Таким образом, в консоль будет выведено сообщение о создании объекта, а затем сообщение о его уничтожении. Please open Telegram to view this post
VIEW IN TELEGRAM
👍6❤3🔥3
rich в Python предоставляет мощные возможности для улучшения визуального оформления терминальных приложений. С помощью rich можно создавать цветные тексты, таблицы, деревья, прогресс-бары и даже подсвечивать код. rich: 1. Цветной текст и стили
from rich import print
# Выводим текст с цветами и стилями
print("[bold red]Ошибка:[/] Неверный ввод!")
print("[green]Успех:[/] Данные сохранены.")
print("Это [italic cyan]пример[/italic cyan] текста с [underline magenta]разными[/underline magenta] стилями.")
2. Форматирование JSON
from rich.console import Console
from rich.json import JSON
console = Console()
# Пример JSON-данных
data = '{"name": "Python3", "type": "Programming Language", "year": 2008}'
# Форматированный вывод JSON
console.print(JSON(data))
3. Создание таблиц
from rich.table import Table
from rich.console import Console
console = Console()
table = Table(title="Сравнение языков программирования")
table.add_column("Язык", style="bold", justify="left")
table.add_column("Скорость", justify="right")
table.add_column("Простота", justify="right")
table.add_row("Python", "Средняя", "Высокая")
table.add_row("Rust", "Высокая", "Средняя")
table.add_row("C++", "Высокая", "Низкая")
console.print(table)
4. Прогресс-бар
from time import sleep
from rich.progress import Progress
with Progress() as progress:
task = progress.add_task("[cyan]Обработка данных...", total=100)
for i in range(100):
sleep(0.05)
progress.update(task, advance=1)
5. Дерево файлов
from rich.tree import Tree
from rich.console import Console
console = Console()
tree = Tree("Проект")
src = tree.add("src")
src.add("main.py")
src.add("utils.py")
tree.add("README.md")
tree.add("requirements.txt")
console.print(tree)
6. Подсветка синтаксиса
from rich.console import Console
from rich.syntax import Syntax
console = Console()
code = '''
def greet(name: str) -> str:
return f"Hello, {name}!"
'''
# Подсвечиваем код
syntax = Syntax(code, "python", theme="monokai", line_numbers=True)
console.print(syntax)
7. Журналирование
from rich.logging import RichHandler
import logging
logging.basicConfig(
level=logging.INFO,
format="%(message)s",
handlers=[RichHandler()]
)
log = logging.getLogger("rich")
log.info("Это [bold green]информация[/bold green].")
log.warning("Это [bold yellow]предупреждение[/bold yellow].")
log.error("Это [bold red]ошибка[/bold red].")
rich выглядит намного приятнее, чем стандартный вывод.Please open Telegram to view this post
VIEW IN TELEGRAM
👍11❤6😍3🔥1
import plotly.graph_objects as go
# Создаем данные для графика
x = [1, 2, 3, 4, 5]
y1 = [10, 11, 12, 13, 14]
y2 = [8, 9, 10, 11, 12]
# Создаем объект Figure
fig = go.Figure()
# Добавляем первый линейный график
fig.add_trace(go.Scatter(x=x, y=y1, mode="lines+markers", name="Линия 1"))
# Добавляем второй линейный график
fig.add_trace(go.Scatter(x=x, y=y2, mode="lines+markers", name="Линия 2"))
# Настраиваем заголовок и подписи осей
fig.update_layout(title="Интерактивный линейный график",
xaxis_title="X",
yaxis_title="Y")
# Отображаем график
fig.show()
# Сохраняем график в HTML-файл
# fig.write_html("my_plotly_chart.html")
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥9👍7❤4🤝1