Python Asyncio Architecture: Event Loops, Tasks, and Futures Explained

Understanding Python’s asyncio architecture is crucial for writing efficient asynchronous code. Here’s a comprehensive guide. Event Loop The event loop is the core of asyncio. It manages and distributes the execution of different tasks. import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Event loop runs the coroutine asyncio.run(main()) Coroutines Coroutines are functions defined with async def. They can be paused and resumed. async def fetch_data(): await asyncio.sleep(1) return "data" # Coroutine object coro = fetch_data() Tasks Tasks wrap coroutines and schedule them on the event loop. ...

July 20, 2022 · 4190 views

Building a Mini Blog with Python and Flask

Learn how to build a simple blog application using Python and Flask. Setup from flask import Flask, render_template, request, redirect, url_for app = Flask(__name__) # Simple in-memory storage posts = [] Routes @app.route('/') def index(): return render_template('index.html', posts=posts) @app.route('/post', methods=['GET', 'POST']) def create_post(): if request.method == 'POST': title = request.form['title'] content = request.form['content'] posts.append({'title': title, 'content': content}) return redirect(url_for('index')) return render_template('create_post.html') Templates <!-- index.html --> {% for post in posts %} <article> <h2>{{ post.title }}</h2> <p>{{ post.content }}</p> </article> {% endfor %} Running flask run Conclusion Build your first Flask blog application! 🚀

June 15, 2022 · 4276 views

Async SQLAlchemy: Best Practices for Async Python Applications

Using SQLAlchemy with async Python requires understanding async patterns. Here’s how. Setup from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) Async Models from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String Base = declarative_base() class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True) name = Column(String) Async Operations async def create_user(name: str): async with AsyncSessionLocal() as session: user = User(name=name) session.add(user) await session.commit() return user async def get_user(user_id: int): async with AsyncSessionLocal() as session: result = await session.get(User, user_id) return result Best Practices Use async context managers Commit transactions explicitly Handle exceptions properly Use connection pooling Close sessions correctly Conclusion Build efficient async database applications with SQLAlchemy! 🐍

March 20, 2022 · 4081 views

Building Your First Rust HTTP API with Axum

Axum is a modern web framework for Rust. Here’s how to build your first API. Setup [dependencies] axum = "0.7" tokio = { version = "1", features = ["full"] } serde = { version = "1.0", features = ["derive"] } Basic Server use axum::{Router, routing::get, Json}; #[tokio::main] async fn main() { let app = Router::new() .route("/", get(handler)); axum::Server::bind(&"0.0.0.0:3000".parse().unwrap()) .serve(app.into_make_service()) .await .unwrap(); } async fn handler() -> Json<serde_json::Value> { Json(serde_json::json!({"message": "Hello, World!"})) } Routes let app = Router::new() .route("/users", get(get_users)) .route("/users/:id", get(get_user)); JSON Handling use serde::{Deserialize, Serialize}; #[derive(Serialize, Deserialize)] struct User { id: u32, name: String, } async fn create_user(Json(user): Json<User>) -> Json<User> { Json(user) } Best Practices Use type-safe routing Handle errors properly Use middleware Test endpoints Document APIs Conclusion Build fast and safe APIs with Rust and Axum! 🦀

November 20, 2021 · 4276 views