Казалось бы, Python — язык высокого уровня с автоматическим управлением памятью. Зачем нам вообще беспокоиться об этом? Но, друзья мои, даже в Пайтоне память не бесконечна. Особенно когда вы работаете с большими данными или создаете высоконагруженные приложения.
memory_profiler — это как швейцарский нож для анализа памяти. Вот пример его использования:
from memory_profiler import profile
@profile
def my_func():
a = [1] * (10 ** 6)
b = [2] * (2 * 10 ** 7)
del b
return a
if __name__ == '__main__':
my_func()
python -m memory_profiler script.py, и вы увидите подробный отчет о использовании памяти. Красота, правда?Если memory_profiler — это швейцарский нож, то line_profiler — это микроскоп. Он покажет вам использование памяти построчно:
@profile
def my_func():
a = [1] * (10 ** 6)
b = [2] * (2 * 10 ** 7)
del b
return a
my_func()
kernprof -l -v script.py, и вы увидите, какая строчка сколько памяти съедает.objgraph — это как рентген для вашего кода. Он позволяет визуализировать объекты в памяти:
import objgraph
x = []
y = [x, [x], dict(x=x)]
objgraph.show_refs([y], filename='sample-graph.png')
А теперь — жемчужина в короне Python 3. tracemalloc — это встроенный модуль для отслеживания выделения памяти:
import tracemalloc
tracemalloc.start()
# ваш код здесь
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics('lineno')
print("[ Топ 10 ]")
for stat in top_stats[:10]:
print(stat)
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥9❤3👍3
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from lineapy import LineaPy
# Создаем синтетические данные
np.random.seed(42)
X = np.random.rand(100, 1) * 10 # 100 случайных точек в диапазоне от 0 до 10
y = 2.5 * X + np.random.randn(100, 1) * 2 # Линейная зависимость с шумом
# Преобразуем данные в DataFrame
data = pd.DataFrame(np.hstack((X, y)), columns=['X', 'y'])
# Создаем модель с помощью LineaPy
model = LineaPy()
model.fit(data['X'], data['y'])
# Предсказания
predictions = model.predict(data['X'])
# Визуализация
plt.scatter(data['X'], data['y'], color='blue', label='Данные')
plt.plot(data['X'], predictions, color='red', label='Предсказание')
plt.xlabel('X')
plt.ylabel('y')
plt.title('Линейная Регрессия с LineaPy')
plt.legend()
plt.show()
Please open Telegram to view this post
VIEW IN TELEGRAM
❤4👍2🤝2
# Установка библиотеки
!pip install fugue
# Импорт необходимых модулей
from fugue import FugueEngine
from fugue.spark import SparkExecutionEngine
# Пример обработки данных
def process_data(df):
return df.groupby("category").agg({"value": "sum"})
# Инициализация движка
engine = SparkExecutionEngine()
# Чтение данных
data = [
{"category": "A", "value": 10},
{"category": "B", "value": 20},
{"category": "A", "value": 30},
]
# Обработка данных
result = engine.run(data, process_data)
# Вывод результата
print(result)
Please open Telegram to view this post
VIEW IN TELEGRAM
👍5❤4🔥3
Монолитная архитектура представляет собой традиционный подход к разработке приложений, где все компоненты тесно связаны и работают как единое целое.
Микросервисная архитектура разбивает приложение на набор небольших, независимых сервисов, каждый из которых отвечает за конкретную бизнес-функцию.
Выбор архитектуры зависит от нескольких факторов:
Выбор между микросервисами и монолитной архитектурой не является универсальным. Каждый подход имеет свои сильные стороны, и правильный выбор зависит от конкретных требований вашего проекта, ресурсов команды и долгосрочных целей. Важно тщательно оценить все факторы и принять решение, которое наилучшим образом соответствует вашим потребностям.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6👍4🔥3
reduce() - это функция, которая применяет указанную функцию к итерируемому объекту, последовательно сводя его к единственному значению. Это мощный инструмент для обработки последовательностей данных, особенно когда нужно выполнить кумулятивные операции.
Пример использования:
from functools import reduce
numbers = [1, 2, 3, 4, 5]
product = reduce(lambda x, y: x * y, numbers)
print(product) # Выведет: 120
@singledispatch позволяет создавать функции, которые ведут себя по-разному в зависимости от типа переданного аргумента. Это элегантная альтернатива множественным условным операторам.
from functools import singledispatch
@singledispatch
def process(arg):
print(f"Обработка объекта: {arg}")
@process.register(int)
def _(arg):
print(f"Обработка целого числа: {arg}")
@process.register(list)
def _(arg):
print(f"Обработка списка длиной {len(arg)}")
process("строка") # Обработка объекта: строка
process(42) # Обработка целого числа: 42
process([1, 2, 3]) # Обработка списка длиной 3
Декоратор @total_ordering значительно упрощает реализацию классов, поддерживающих операции упорядочивания. Достаточно определить методы eq() и один из методов сравнения (lt, le, gt или ge), а остальные будут автоматически созданы.
from functools import total_ordering
@total_ordering
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def __eq__(self, other):
return self.age == other.age
def __lt__(self, other):
return self.age < other.age
p1 = Person("Алиса", 25)
p2 = Person("Боб", 30)
print(p1 < p2) # True
print(p1 <= p2) # True
print(p1 > p2) # False
print(p1 >= p2) # False
Please open Telegram to view this post
VIEW IN TELEGRAM
👍6❤3🔥2⚡1
Функция cache предоставляет простой способ кэширования результатов функции без ограничения размера кэша. Это может быть полезно, когда вы уверены, что количество уникальных входных данных ограничено.
from functools import cache
@cache
def fibonacci(n):
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)
print(fibonacci(100)) # Мгновенный результат даже для больших чисел
Декоратор @wraps помогает сохранить метаданные оригинальной функции при создании декораторов. Это особенно важно при использовании инструментов документации и отладки.
from functools import wraps
def my_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
"""Это документация обертки"""
print("До вызова функции")
result = func(*args, **kwargs)
print("После вызова функции")
return result
return wrapper
@my_decorator
def say_hello(name):
"""Приветствует пользователя по имени"""
print(f"Привет, {name}!")
say_hello("Мария")
print(say_hello.__name__) # Выведет: say_hello
print(say_hello.__doc__) # Выведет: Приветствует пользователя по имени
Please open Telegram to view this post
VIEW IN TELEGRAM
👍4❤3🔥2
Функция type() в Python - это не просто инструмент для определения типа объекта. Она также может быть использована для создания новых классов динамически. Вот простой пример:
MyClass = type('MyClass', (), {'x': 42, 'my_method': lambda self: print("Hello!")})
obj = MyClass()
print(obj.x) # Выведет: 42
obj.my_method() # Выведет: Hello!Метаклассы - это классы классов. Они позволяют нам контролировать процесс создания классов. Рассмотрим пример:
class MyMetaclass(type):
def __new__(cls, name, bases, attrs):
attrs['additional_method'] = lambda self: print("I'm additional!")
return super().__new__(cls, name, bases, attrs)
class MyClass(metaclass=MyMetaclass):
pass
obj = MyClass()
obj.additional_method() # Выведет: I'm additional!
Динамическое создание классов и функций может быть полезно в различных сценариях:
- Фабрики классов: создание классов на основе внешних данных или конфигурации.
- Декораторы классов: модификация классов без изменения их исходного кода.
- ORM (Object-Relational Mapping): динамическое создание классов на основе структуры базы данных.
Please open Telegram to view this post
VIEW IN TELEGRAM
👍5❤4🔥3
Функциональное программирование - это парадигма, в которой процесс вычисления рассматривается как вычисление математических функций. Основная идея заключается в том, чтобы избегать изменяемого состояния и мутаций данных.
Функциональное программирование становится все популярнее благодаря своим преимуществам в читаемости, тестируемости и параллельном выполнении. Хотя оно может потребовать некоторого времени для освоения, инвестиции в изучение ФП могут значительно улучшить качество вашего кода и эффективность разработки.
Please open Telegram to view this post
VIEW IN TELEGRAM
❤5👍3🔥2
👍7🤔7🤨3
Таким образом, пока а<4, у нас просто происходит увеличение этой переменной, а из-за условия, итерация (и код ниже) скипаются.
Далее у нас а станет равно 4. У нас не сработает ни один из if'ов и мы добавим это число 4 в result.
Ну и на след. итерации а == 5, у нас сработает break, который принудительно завершит цикл в той же строке.
Ответ: 4
Please open Telegram to view this post
VIEW IN TELEGRAM
👍13❤5🔥3
Представьте, что ваше приложение – это кухня в популярном ресторане. В синхронном мире у вас один шеф-повар, готовящий блюда по очереди. С asyncio у вас целая команда виртуозов, жонглирующих сковородками и готовящих несколько блюд одновременно. Вот это производительность!
import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
loop = asyncio.get_event_loop()
loop.set_default_executor(concurrent.futures.ThreadPoolExecutor(max_workers=10))
sem = asyncio.Semaphore(10)
async def controlled_task(i):
async with sem:
# Ваш асинхронный код здесь
await asyncio.sleep(1)
print(f"Задача {i} выполнена!")
class AsyncContextManager:
async def __aenter__(self):
print("Entering the matrix...")
await asyncio.sleep(1)
return self
async def __aexit__(self, exc_type, exc, tb):
print("Exiting the matrix...")
await asyncio.sleep(1)
async def main():
async with AsyncContextManager() as manager:
print("We're in!")
async def async_range(start, stop):
for i in range(start, stop):
await asyncio.sleep(0.1)
yield i
async def main():
async for num in async_range(0, 10):
print(num)
Asyncio – это не просто библиотека, это образ мышления. Освоив его, вы сможете создавать приложения, которые будут работать быстрее, эффективнее и элегантнее. Помните: в мире асинхронного программирования нет ничего невозможного!
Please open Telegram to view this post
VIEW IN TELEGRAM
👍5🔥4❤3
Представьте, что ваш код – это оркестр в Мариинском театре. Каждый инструмент (поток или корутина) играет свою партию. А вы – дирижёр, который должен убедиться, что все звучит гармонично. Вот только как это сделать, когда все играют одновременно?
➡️ 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