Разбирает массивы, списки, стеки, очереди, деревья и графы, а также показывает, как применять их для оптимизации и построения эффективных алгоритмов. Помогает закрепить знания через практические задачи и реальные кейсы программирования.
tags: #обучение #python
Please open Telegram to view this post
VIEW IN TELEGRAM
❤5👍3🔥2👨💻1
Разбирает создание сервисов через asyncio и Starlette, взаимодействие компонентов через RabbitMQ и валидацию данных с помощью Pydantic. Показывает тестирование микросервисов с pytest и разработку полноценного чата между пользователем и сервисной архитектурой.
tags: #обучение #python
Please open Telegram to view this post
VIEW IN TELEGRAM
👍6❤3👨💻3🔥1
sleep() создают дополнительную нагрузку и могут только усилить проблему. Более надёжный подход — использовать экспоненциальную задержку с ограничением количества попыток.import time
import random
def retry(func, attempts=5, delay=1):
for attempt in range(attempts):
try:
return func()
except Exception as error:
if attempt == attempts - 1:
raise error
wait = delay * (2 ** attempt)
wait += random.uniform(0, 1)
print(f"retry after {wait:.2f}s")
time.sleep(wait)
def request():
if random.random() < 0.7:
raise RuntimeError("temporary error")
return "success"
print(retry(request))
• После каждой ошибки время ожидания увеличивается: 1 → 2 → 4 → 8 секунд, снижая нагрузку на внешний сервис.
• Случайная задержка (
jitter) добавляется, чтобы несколько параллельных процессов не повторяли запрос одновременно.• После превышения количества попыток исключение возвращается выше по стеку для дальнейшей обработки.
tags: #python #автоматизация #разработка
Please open Telegram to view this post
VIEW IN TELEGRAM
❤7👍4⚡2👨💻1
В программе разбираются управление проектами разного масштаба, разработка веб-приложений на Django, Flask и других фреймворках, анализ данных, рефакторинг и поиск проблем в коде. Также изучаются Git, GitHub, визуальная отладка, модульное тестирование с pytest и создание собственных Python-пакетов.
tags: #обучение #python #разработка
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥3👨💻3❤2👍2
mtime) ненадёжно: файл может быть перезаписан без изменения содержимого или изменён с сохранением даты. Более точный подход — вычислять хеш содержимого и сравнивать реальные изменения данных.import hashlib
import time
from pathlib import Path
def file_hash(path):
sha = hashlib.sha256()
with open(path, "rb") as file:
for chunk in iter(lambda: file.read(8192), b""):
sha.update(chunk)
return sha.hexdigest()
target = Path("/etc/nginx/nginx.conf")
old_hash = file_hash(target)
while True:
new_hash = file_hash(target)
if new_hash != old_hash:
print("file changed:", target)
old_hash = new_hash
time.sleep(10)
• Файл читается небольшими блоками, поэтому метод подходит даже для больших объектов, которые нельзя полностью загрузить в память.
• SHA-256 создаёт уникальный отпечаток содержимого, поэтому изменение данных определяется независимо от времени изменения файла.
• Такой механизм можно использовать как основу для автоматического контроля конфигураций, синхронизации и запуска действий после изменения данных.
tags: #python #автоматизация #разработка
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6👍4👨💻3🔥2
В программе разбирается создание Django-проекта, анализ SQL-запросов приложения, устройство middleware и разработка собственного компонента с базовой функциональностью. В конце курса участники создадут полноценный пакет для Django и научатся публиковать его в Python Package Index.
tags: #обучение #python #django
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥4❤3👍3👨💻3
asyncio.Queue позволяет разделить генерацию и обработку задач, ограничить параллельность и построить управляемый конвейер.import asyncio
async def worker(queue):
while True:
task = await queue.get()
try:
print("processing:", task)
await asyncio.sleep(1)
finally:
queue.task_done()
async def main():
queue = asyncio.Queue(maxsize=100)
workers = [
asyncio.create_task(worker(queue))
for _ in range(3)
]
for i in range(10):
await queue.put(f"job-{i}")
await queue.join()
for worker_task in workers:
worker_task.cancel()
asyncio.run(main())
•
asyncio.Queue создаёт буфер между производителями и обработчиками, поэтому скорость поступления задач не ломает обработку.•
maxsize ограничивает количество ожидающих задач и создаёт естественный механизм обратного давления.• Несколько воркеров позволяют выполнять операции параллельно, сохраняя полный контроль над количеством одновременно работающих задач.
tags: #python #разработка
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6🔥5👍4👨💻2
В программе разбираются синтаксис и структуры данных Python, объектно-ориентированное программирование и модули. Также изучается практическая работа с файлами, CSV, JSON, PDF и ZIP, чтобы полученные знания можно было применять при разработке реальных программ.
tags: #обучение #python
Please open Telegram to view this post
VIEW IN TELEGRAM
❤6🔥5👍2👨💻1
rsync, ssh, kubectl, terraform, docker. Проблема начинается, когда одна команда зависает или одновременно запускается слишком много процессов. asyncio позволяет управлять десятками внешних команд, ограничивать параллельность и автоматически завершать зависшие операции.import asyncio
async def run(cmd, sem):
async with sem:
proc = await asyncio.create_subprocess_exec(
*cmd, stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
try:
out, err = await asyncio.wait_for(
proc.communicate(), timeout=30
)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise
return proc.returncode, out, err
async def main():
sem = asyncio.Semaphore(5)
jobs = [
["ssh", host, "uptime"]
for host in ["server1", "server2", "server3"]
]
results = await asyncio.gather(
*(run(job, sem) for job in jobs)
)
for code, out, err in results:
print(code, out.decode().strip())
asyncio.run(main())
•
Semaphore(5) ограничивает количество одновременно работающих процессов, поэтому автоматизация не создаёт неконтролируемую нагрузку на систему или сеть.•
wait_for() устанавливает таймаут для каждой команды. Если внешний процесс зависает, Python принудительно завершает его и освобождает воркер.•
asyncio.gather() позволяет собрать результаты множества операций в одной точке, что удобно для массового выполнения команд на серверах.ssh, kubectl, docker или любые другие CLI-команды, сохраняя контроль над количеством процессов и временем выполнения каждой операции.tags: #python #автоматизация #разработка
Please open Telegram to view this post
VIEW IN TELEGRAM
❤8👍6👨💻3⚡1
Вы освоите синтаксис и структуру Python, библиотеки и инструменты разработки, а также Django для создания веб-приложений. В ходе обучения соберёте портфолио проектов и подготовитесь к работе на фрилансе.
tags: #обучение #python
Please open Telegram to view this post
VIEW IN TELEGRAM
❤4👍2🔥2👨💻1
import sqlite3
import time
db = sqlite3.connect("jobs.db", isolation_level=None)
db.execute("""
CREATE TABLE IF NOT EXISTS jobs (
id INTEGER PRIMARY KEY,
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
created_at REAL NOT NULL
)
""")
db.execute("""
INSERT INTO jobs (payload, created_at)
VALUES (?, ?)
""", ("deploy server-01", time.time()))
while True:
with db:
job = db.execute("""
SELECT id, payload
FROM jobs
WHERE status = 'pending'
ORDER BY id
LIMIT 1
""").fetchone()
if not job:
break
db.execute(
"UPDATE jobs SET status = 'running' WHERE id = ?",
(job[0],)
)
try:
print(f"Выполняем: {job[1]}")
# здесь выполняется реальная задача
time.sleep(2)
db.execute(
"UPDATE jobs SET status = 'done' WHERE id = ?",
(job[0],)
)
except Exception:
db.execute(
"UPDATE jobs SET status = 'failed' WHERE id = ?",
(job[0],)
)
• Задача сначала записывается на диск, поэтому перезапуск процесса не уничтожает очередь.
• Состояния
pending, running, done и failed позволяют восстановить состояние автоматизации после сбоя.• SQLite поддерживает транзакции, блокировки и WAL, поэтому для небольших локальных job-очередей отдельный брокер сообщений может быть избыточен.
tags: #python #автоматизация #разработка
Please open Telegram to view this post
VIEW IN TELEGRAM
🔥5❤4⚡3👨💻2👍1