Contenidos
Últimas Novedades sobre Hilos, Procesos, Red y Seguridad en Python
Hilos y Procesos
Hilos con concurrent.futures
El módulo concurrent.futures proporciona una interfaz de alto nivel para ejecutar tareas de forma asíncrona usando hilos o procesos.
|
1 2 3 4 5 6 7 8 9 10 11 |
from concurrent.futures import ThreadPoolExecutor def tarea(n): print(f"Tarea {n} en ejecución") return n * n with ThreadPoolExecutor(max_workers=5) as executor: futuros = [executor.submit(tarea, i) for i in range(5)] resultados = [f.result() for f in futuros] print(f"Resultados: {resultados}") |
Procesos con concurrent.futures
El mismo módulo también puede ser usado para manejar procesos en paralelo.
|
1 2 3 4 5 6 7 8 9 10 11 |
from concurrent.futures import ProcessPoolExecutor def tarea_proceso(n): print(f"Tarea {n} en ejecución en un proceso separado") return n * n with ProcessPoolExecutor(max_workers=5) as executor: futuros = [executor.submit(tarea_proceso, i) for i in range(5)] resultados = [f.result() for f in futuros] print(f"Resultados: {resultados}") |
Red
Comunicación HTTP con HTTPX
HTTPX es una biblioteca HTTP moderna para Python que permite realizar peticiones HTTP/1.1 y HTTP/2 de forma asíncrona.
|
1 2 3 4 5 6 7 8 9 10 11 |
import httpx async def fetch(url): async with httpx.AsyncClient() as client: response = await client.get(url) return response.text import asyncio url = "https://www.example.com" contenido = asyncio.run(fetch(url)) print(contenido) |
Websockets con websockets
La biblioteca websockets permite la creación de clientes y servidores WebSocket, facilitando la comunicación en tiempo real.
|
1 2 3 4 5 6 7 8 9 10 11 |
import asyncio import websockets async def echo(websocket, path): async for message in websocket: await websocket.send(message) start_server = websockets.serve(echo, "localhost", 8765) asyncio.get_event_loop().run_until_complete(start_server) asyncio.get_event_loop().run_forever() |
Seguridad
Autenticación con Authlib
Authlib es una biblioteca que facilita la implementación de autenticación OAuth y OpenID Connect en aplicaciones Python.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
from authlib.integrations.flask_client import OAuth from flask import Flask, redirect, url_for app = Flask(__name__) app.secret_key = 'secret' oauth = OAuth(app) oauth.register('example', client_id='EXAMPLE_CLIENT_ID', client_secret='EXAMPLE_CLIENT_SECRET', authorize_url='https://example.com/authorize', access_token_url='https://example.com/token', client_kwargs={'scope': 'openid profile email'}) @app.route('/') def homepage(): return 'Welcome to the secure app' @app.route('/login') def login(): redirect_uri = url_for('auth', _external=True) return oauth.example.authorize_redirect(redirect_uri) @app.route('/auth') def auth(): token = oauth.example.authorize_access_token() user = oauth.example.parse_id_token(token) return f'Hello, {user["name"]}!' if __name__ == '__main__': app.run() |
Seguridad en API con FastAPI
FastAPI es un framework moderno y de alto rendimiento para construir APIs con Python, que incluye herramientas para gestionar la autenticación y la autorización.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm app = FastAPI() oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") @app.post("/token") async def token(form_data: OAuth2PasswordRequestForm = Depends()): if form_data.username == "user" and form_data.password == "password": return {"access_token": form_data.username, "token_type": "bearer"} raise HTTPException(status_code=400, detail="Incorrect username or password") @app.get("/users/me") async def read_users_me(token: str = Depends(oauth2_scheme)): if token != "user": raise HTTPException(status_code=401, detail="Invalid token") return {"username": token} if __name__ == '__main__': import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000) |