evalship-hq / ponytail

Output

# Reading Temperature from a Thermistor via MCP3008 Here's a Python function to read temperature from a thermistor connected to an MCP3008 ADC on a Raspberry Pi. ## Prerequisites ```bash pip install adafruit-circuitpython-mcp3xxx ``` Enable SPI on your Pi via `sudo raspi-config` → Interface Options → SPI. ## Circuit Assumptions A common voltage divider setup: ``` 3.3V ---[ Series Resistor R ]---+--- MCP3008 CH0 | [ Thermistor ] | GND ``` ## Code ```python import math import board import busio import digitalio import adafruit_mcp3xxx.mcp3008 as MCP from adafruit_mcp3xxx.analog_in import AnalogIn def setup_mcp3008(channel=0): """Initialize MCP3008 and return the requested analog input channel.""" spi = busio.SPI(clock=board.SCK, MISO=board.MISO, MOSI=board.MOSI) cs = digitalio.DigitalInOut(board.D5) # Chip-select pin (adjust as needed) mcp = MCP.MCP3008(spi, cs) channels = [MCP.P0, MCP.P1, MCP.P2, MCP.P3, MCP.P4, MCP.P5, MCP.P6, MCP.P7] return AnalogIn(mcp, channels[channel]) def read_temperature_celsius( adc_channel, v_supply=3.3, # Supply voltage (V) series_resistor=10000.0, # Series resistor in the divider (ohms) nominal_resistance=10000.0, # Thermistor resistance at nominal temp (ohms) nominal_temp=25.0, # Nominal temperature (°C) beta=3950.0, # Beta coefficient of the thermistor thermistor_on_ground=True, # True if thermistor is between ADC node and GND ): """ Read temperature in Celsius from a thermistor using the Beta (Steinhart simplified) equation. Parameters ---------- adc_channel : AnalogIn The MCP3008 analog input channel object. v_supply : float The voltage feeding the divider. series_resistor : float The fixed resistor value in ohms. nominal_resistance : float Thermistor resistance at the nominal temperature. nominal_temp : float Nominal temperature in Celsius (usually 25°C). beta : float The thermistor's Beta coefficient (from its datasheet). thermistor_on_ground : bool Orientation of the thermistor in the divider. Returns ------- float Temperature in degrees Celsius. """ # MCP3008 is 10-bit (0-1023); voltage read via AnalogIn voltage = adc_channel.voltage # already scaled to volts by the library # Avoid division by zero / invalid readings if voltage <= 0 or voltage >= v_supply: raise ValueError(f"Invalid ADC voltage reading: {voltage:.3f} V") # Calculate thermistor resistance from the voltage divider if thermistor_on_ground: # V_adc = V_supply * R_thermistor / (R_series + R_thermistor) r_thermistor = series_resistor * voltage / (v_supply - voltage) else: # Thermistor is on the high side r_thermistor = series_resistor * (v_supply - voltage) / voltage # Beta equation (simplified Steinhart-Hart) # 1/T = 1/T0 + (1/Beta) * ln(R/R0) (T in Kelvin) t0_kelvin = nominal_temp + 273.15 inv_t = (1.0 / t0_kelvin) + (1.0 / beta) * math.log(r_thermistor / nominal_resistance) temp_kelvin = 1.0 / inv_t return temp_kelvin - 273.15 # --------------------------------------------------------------------------- # Example usage # --------------------------------------------------------------------------- if __name__ == "__main__": import time ch0 = setup_mcp3008(channel=0) try: while True: temp_c = read_temperature_celsius( ch0, v_supply=3.3, series_resistor=10000.0, nominal_resistance=10000.0, nominal_temp=25.0, beta=3950.0, ) print(f"Temperature: {temp_c:.2f} °C") time.sleep(1.0) except KeyboardInterrupt: print("\nStopped.") ``` ## Key Notes 1. **Thermistor parameters** — Check your ther…

Output

```python from gpiozero import MCP3008 from math import log # Vishay NTCLE100E3103 / generic 10k NTC, Beta model R_FIXED = 10_000 # series resistor (ohms) R0 = 10_000 # thermistor resistance at 25 °C BETA = 3950 # thermistor beta (check your datasheet) T0 = 298.15 # 25 °C in kelvin def read_celsius(channel=0): adc = MCP3008(channel=channel) v = adc.value # 0..1 fraction of Vref if not 0 < v < 1: raise ValueError(f"ADC reading {v} out of range (open/short circuit?)") r = R_FIXED * v / (1 - v) # thermistor resistance, divider: 3V3 - R_FIXED - node - thermistor - GND kelvin = 1 / (1 / T0 + log(r / R0) / BETA) return kelvin - 273.15 ``` ```python # self-check: at R == R0 the reading must be 25 °C def _check(): r, kelvin = R0, 1 / (1 / T0 + log(R0 / R0) / BETA) assert abs(kelvin - 273.15 - 25.0) < 0.01 _check() ``` Skipped: averaging/smoothing and config flags — add if your readings are noisy. **You must set `BETA`, `R0`, `R_FIXED` and the divider wiring to match your hardware; wrong values give wrong temps silently.** Steinhart–Hart (3 coefficients) beats Beta if you need accuracy over a wide range. Didn't run it (no Pi/gpiozero here).

Output

# Duration String Parser Here's a Python function that parses duration strings like `'1h30m45s'` into total seconds: ```python import re def parse_duration(duration: str) -> int: """ Parse a duration string into total seconds. Supports units: d (days), h (hours), m (minutes), s (seconds). Example: '1h30m45s' -> 5445 Args: duration: A duration string like '1h30m45s', '2d', '90m'. Returns: Total number of seconds as an integer. Raises: ValueError: If the string is empty or contains invalid format. """ if not duration or not duration.strip(): raise ValueError("Duration string cannot be empty") duration = duration.strip().lower() # Multipliers for each unit (in seconds) multipliers = { 'd': 86400, 'h': 3600, 'm': 60, 's': 1, } # Match groups of digits followed by a unit character pattern = re.compile(r'(\d+)([dhms])') total = 0 matched_length = 0 for match in pattern.finditer(duration): value, unit = match.groups() total += int(value) * multipliers[unit] matched_length += len(match.group(0)) # Ensure the entire string was consumed (no invalid characters) if matched_length != len(duration): raise ValueError(f"Invalid duration format: {duration!r}") return total ``` ## Usage Examples ```python print(parse_duration('1h30m45s')) # 5445 print(parse_duration('2d')) # 172800 print(parse_duration('90m')) # 5400 print(parse_duration('45s')) # 45 print(parse_duration('1d2h3m4s')) # 93784 # Errors try: parse_duration('abc') except ValueError as e: print(e) # Invalid duration format: 'abc' ``` ## How It Works 1. **Validation**: Rejects empty/whitespace-only strings. 2. **Regex matching**: `(\d+)([dhms])` captures a number paired with a unit. 3. **Accumulation**: Each `value × multiplier` is summed into the total. 4. **Integrity check**: Comparing `matched_length` to the full string length ensures there are no stray or invalid characters (e.g., `'1h30x'` fails). ## Optional: Support for Floats If you want to allow fractional values (e.g., `'1.5h'`): ```python def parse_duration_float(duration: str) -> float: if not duration or not duration.strip(): raise ValueError("Duration string cannot be empty") duration = duration.strip().lower() multipliers = {'d': 86400, 'h': 3600, 'm': 60, 's': 1} pattern = re.compile(r'(\d+(?:\.\d+)?)([dhms])') total = 0.0 matched_length = 0 for match in pattern.finditer(duration): value, unit = match.groups() total += float(value) * multipliers[unit] matched_length += len(match.group(0)) if matched_length != len(duration): raise ValueError(f"Invalid duration format: {duration!r}") return total print(parse_duration_float('1.5h')) # 5400.0 ``` Let me know if you need support for additional units (weeks, milliseconds) or different formats!

Output

## Email Validation Function Here's a robust email validation function with multiple levels of checking: ```python import re def validate_email(email: str) -> dict[str, bool | str]: """ Validates an email address with multiple checks. Args: email: The email address to validate. Returns: A dictionary with: - 'is_valid': Boolean indicating if the email is valid. - 'message': A descriptive message about the validation result. """ # Check if email is a non-empty string if not isinstance(email, str) or not email.strip(): return {"is_valid": False, "message": "Email must be a non-empty string."} email = email.strip() # Check overall length (max 254 characters per RFC 5321) if len(email) > 254: return {"is_valid": False, "message": "Email exceeds maximum length of 254 characters."} # Check for exactly one '@' symbol if email.count("@") != 1: return {"is_valid": False, "message": "Email must contain exactly one '@' symbol."} local, domain = email.split("@") # --- Local part checks --- if not local: return {"is_valid": False, "message": "Local part (before '@') cannot be empty."} if len(local) > 64: return {"is_valid": False, "message": "Local part exceeds maximum length of 64 characters."} if local.startswith(".") or local.endswith("."): return {"is_valid": False, "message": "Local part cannot start or end with a dot."} if ".." in local: return {"is_valid": False, "message": "Local part cannot contain consecutive dots."} local_pattern = r'^[a-zA-Z0-9._%+\-]+$' if not re.match(local_pattern, local): return {"is_valid": False, "message": "Local part contains invalid characters."} # --- Domain part checks --- if not domain: return {"is_valid": False, "message": "Domain part (after '@') cannot be empty."} if domain.startswith("-") or domain.endswith("-"): return {"is_valid": False, "message": "Domain cannot start or end with a hyphen."} if ".." in domain: return {"is_valid": False, "message": "Domain cannot contain consecutive dots."} if "." not in domain: return {"is_valid": False, "message": "Domain must contain at least one dot."} domain_pattern = r'^[a-zA-Z0-9.\-]+$' if not re.match(domain_pattern, domain): return {"is_valid": False, "message": "Domain contains invalid characters."} # Check TLD (last part after the final dot) tld = domain.split(".")[-1] if len(tld) < 2: return {"is_valid": False, "message": "Top-level domain must be at least 2 characters."} if not tld.isalpha(): return {"is_valid": False, "message": "Top-level domain must contain only letters."} return {"is_valid": True, "message": "Email is valid."} def is_valid_email(email: str) -> bool: """Simplified wrapper that returns only a boolean.""" return validate_email(email)["is_valid"] ``` --- ### Example Usage ```python test_emails = [ "user@example.com", # Valid "user.name+tag@domain.co", # Valid "invalid-email", # No @ symbol "@nodomain.com", # Empty local part "user@", # Empty domain "user@@domain.com", # Multiple @ symbols "user@domain", # No TLD "user@domain.c", # TLD too short ".user@domain.com", # Local starts with dot "user..name@domain.com", # Consecutive dots "a" * 65 + "@domain.com", # Local part too long ] for email in test_emails: result = validate_email(email) status = "✅" if result["is_valid"] else "❌" print(f"{status} {email!r}: {result['message']}") ``` **Output:** ``` ✅ 'user@example.com': Email is valid. ✅ 'user.name+tag@domain.co': Email is valid. ❌ 'invalid-email': Email must contain exactly one '@' symbol. ❌ '@nodomain.com': Local part (before '@') cannot be empty. ❌ 'user@': Domain part (after '@') cannot be empty. ❌ 'user@@domain.com': Email m…

Output

Don't write one. Use this: ```python from email_validator import validate_email def validate_email_address(email): try: validate_email(email) return True except: return False ``` Install it: `pip install email_validator` **Why:** The RFC for email is 300+ pages. Hand-rolled regex fails on real addresses (subdomains, plus addressing, quoted strings). This library handles the hard cases and is already in most projects. **Skipped:** custom regex, edge case handling, DNS verification. **Risk:** if you need to reject addresses your mail provider won't accept, test against your actual provider first—this validator is strict-correct, not provider-matched.

Output

```python import re def parse_duration(s): units = {'h': 3600, 'm': 60, 's': 1} total, matched = 0, re.findall(r'(\d+)([hms])', s) if not matched or ''.join(f'{n}{u}' for n, u in matched) != s: raise ValueError(f'invalid duration: {s!r}') for n, u in matched: total += int(n) * units[u] return total assert parse_duration('1h30m45s') == 5445 assert parse_duration('45s') == 45 assert parse_duration('2h') == 7200 for bad in ('', '1x', '1h junk', '30'): try: parse_duration(bad); assert False except ValueError: pass ``` The reassembly check (`''.join(...) != s`) rejects garbage/duplicate/out-of-order units without a stricter regex. Skipped: fractional (`1.5h`), days/weeks, negative values — none were asked for. Risk: units must appear in h→m→s order, matching the example.

Output

# Refactored Code ```python def double_positive_amounts(records): """Return a list of doubled 'amount' values for records with a positive amount. Args: records: An iterable of dict-like objects, each potentially containing an 'amount' key with a numeric value. Returns: A list where each positive 'amount' has been multiplied by 2. """ doubled_amounts = [] for record in records: amount = record.get('amount') if amount is not None and amount > 0: doubled_amounts.append(amount * 2) return doubled_amounts ``` Or, if you prefer a more concise, Pythonic version: ```python def double_positive_amounts(records): """Return doubled 'amount' values for records with a positive amount.""" return [ record['amount'] * 2 for record in records if record.get('amount') is not None and record['amount'] > 0 ] ``` --- # Step-by-Step Write-Up of Changes ### 1. Renamed the function `p` → `double_positive_amounts` **Why:** `p` is meaningless. A function name should describe *what it does*. The new name communicates the behavior (doubling) and the filtering condition (positive amounts) at a glance, so callers don't have to read the body to understand it. ### 2. Renamed the parameter `d` → `records` **Why:** `d` gives no hint about what's being passed in. Since we iterate over it and treat each item as a dict, `records` (a plural noun) signals that it's a collection of data items. ### 3. Renamed the loop variable `x` → `record` **Why:** `x` is generic. `record` matches the `records` collection name, making the relationship between the collection and its elements obvious and self-documenting. ### 4. Renamed the result list `r` → `doubled_amounts` **Why:** `r` doesn't convey intent. `doubled_amounts` tells the reader exactly what the list accumulates, which clarifies the function's return value. ### 5. Renamed the dict key `'a'` → `'amount'` **Why:** `'a'` is a cryptic key. Assuming it represents a monetary or quantity value, `'amount'` makes the data's meaning explicit. *(Note: this change assumes you control the data source. If the actual key is genuinely `'a'`, you must keep it — I flagged this so you can verify.)* ### 6. Extracted `record.get('amount')` into a local variable `amount` **Why:** The original code called `x.get('a')` and then accessed `x['a']` separately — two lookups for the same value. Extracting it to `amount`: - Avoids the redundant second dictionary lookup (minor performance gain). - Avoids a subtle inconsistency where `.get()` (safe) and `['a']` (raises `KeyError`) were mixed. - Makes the condition and the computation easier to read. ### 7. Changed the truthiness check `if x.get('a')` → `if amount is not None` **Why:** This is the most important correctness change. The original `if x.get('a') and x['a'] > 0` relied on truthiness. The problem: - `x.get('a')` returns `None` if the key is missing — that part is fine. - But a value of `0` is falsy, so `0` would short-circuit and be excluded. That's *accidentally* OK here because `0 > 0` is already false, **but** the intent was unclear. Using `amount is not None` makes the "key may be missing" guard explicit and separate from the "must be positive" business rule (`amount > 0`). The two conditions now read as two distinct checks rather than one entangled truthiness trick. ### 8. Added a docstring **Why:** Documents the function's purpose, expected input shape, and return value, so future readers (and tooling/IDEs) understand the contract without reverse-engineering the logic. --- # One Caveat / Assumption to Verify I renamed the key `'a'` to `'amount'` **for readability**, but this changes behavior if your real data actually uses `'a'`. Before adopting that change: - If you own the data model, rename the key everywhere consistently. - If the key is fixed as `'a'` by an external source, keep `'a'` but still apply all the other improvements (variable names, docstring, `is not None` check). If `None`/`0`/negative handling matter…

Output

```python def p(d): return [x['a'] * 2 for x in d if x.get('a', 0) > 0] ``` What changed and why: - Folded the loop into a list comprehension — same result, one line, standard Python idiom. - `x.get('a') and x['a'] > 0` → `x.get('a', 0) > 0`. The original guards against missing/falsy `a` then compares; a default of `0` does both in one check. Behaves identically for your case (positive numbers), and `0`/`None`/missing all fail the `> 0` test. Left names (`p`, `d`, `x`, `a`) alone — I don't know what they mean, and renaming blind is how you break callers. Rename them yourself if you have the context. Skipped: tests (trivial change, behavior preserved) and renaming. One risk: if `a` can be a non-number (string, etc.), the original `and` short-circuit and the new `.get('a', 0) > 0` both raise on the `> 0` compare — no behavior change, just noting it.

Output

# Rate Limiting in FastAPI Here are several approaches, from simplest to most robust: ## 1. **Using `slowapi` (Recommended - Easiest)** ```bash pip install slowapi ``` ```python from fastapi import FastAPI, Request from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from slowapi.middleware import SlowAPIMiddleware app = FastAPI() limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware(SlowAPIMiddleware) async def _rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded): return JSONResponse( status_code=429, content={"detail": "Rate limit exceeded"}, ) @app.get("/api/data") @limiter.limit("5/minute") # 5 requests per minute async def get_data(): return {"message": "Success"} @app.post("/api/submit") @limiter.limit("10/hour") # 10 requests per hour async def submit_data(data: dict): return {"status": "received"} ``` **Rate limit formats:** - `"5/minute"` - 5 requests per minute - `"100/hour"` - 100 requests per hour - `"1000/day"` - 1000 requests per day - `"5/second"` - 5 requests per second --- ## 2. **Manual Implementation with In-Memory Store** Simple but doesn't persist across restarts: ```python from fastapi import FastAPI, HTTPException, Request from datetime import datetime, timedelta from collections import defaultdict import asyncio app = FastAPI() # Store: {client_ip: [(timestamp, count)]} request_store = defaultdict(list) async def check_rate_limit(request: Request, max_requests: int, window_seconds: int) -> bool: client_ip = request.client.host now = datetime.now() cutoff_time = now - timedelta(seconds=window_seconds) # Clean old requests request_store[client_ip] = [ req_time for req_time in request_store[client_ip] if req_time > cutoff_time ] # Check limit if len(request_store[client_ip]) >= max_requests: return False request_store[client_ip].append(now) return True @app.get("/api/data") async def get_data(request: Request): if not await check_rate_limit(request, max_requests=5, window_seconds=60): raise HTTPException(status_code=429, detail="Rate limit exceeded") return {"message": "Success"} ``` --- ## 3. **Using Redis (Production-Ready)** For distributed systems: ```bash pip install slowapi redis ``` ```python from fastapi import FastAPI from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.middleware import SlowAPIMiddleware import redis app = FastAPI() # Connect to Redis redis_client = redis.Redis(host='localhost', port=6379, db=0) limiter = Limiter( key_func=get_remote_address, storage_uri="redis://localhost:6379" ) app.state.limiter = limiter app.add_middleware(SlowAPIMiddleware) @app.get("/api/data") @limiter.limit("5/minute") async def get_data(): return {"message": "Success"} ``` --- ## 4. **Advanced: Custom Rate Limiter with Dependencies** ```python from fastapi import FastAPI, Depends, HTTPException, Request from datetime import datetime, timedelta from typing import Optional import asyncio app = FastAPI() class RateLimiter: def __init__(self, max_requests: int, window_seconds: int): self.max_requests = max_requests self.window_seconds = window_seconds self.requests = {} async def __call__(self, request: Request): client_ip = request.client.host now = datetime.now() if client_ip not in self.requests: self.requests[client_ip] = [] # Clean old requests self.requests[client_ip] = [ req_time for req_time in self.requests[client_ip] if (now - req_time).seconds < self.window_seconds ] # Check limit if len(self.requests[client_ip]) >= self.max_requests: raise HTTPException( status_code=429, detail=f"Rate lim…

Output

# Rate Limiting in FastAPI The most popular solution is **SlowAPI** (built on the `limits` library). Here's how to set it up. ## Installation ```bash pip install slowapi ``` ## Basic Setup ```python from fastapi import FastAPI, Request from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded # Create limiter keyed by client IP limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @app.get("/api/data") @limiter.limit("5/minute") # max 5 requests per minute per IP async def get_data(request: Request): # 'request' param is REQUIRED return {"message": "Here's your data"} ``` > ⚠️ **Important:** The endpoint **must** have a `request: Request` parameter, or SlowAPI will raise an error. ## Rate Limit String Formats ```python @limiter.limit("100/day") @limiter.limit("10/minute") @limiter.limit("5/second") @limiter.limit("1000/hour;100/minute") # multiple limits at once ``` ## Applying Different Limits per Endpoint ```python @app.post("/login") @limiter.limit("5/minute") # stricter for sensitive endpoints async def login(request: Request): ... @app.get("/public") @limiter.limit("60/minute") async def public(request: Request): ... ``` ## Keying by User Instead of IP If users are authenticated, limiting by user ID is fairer (avoids penalizing users behind shared IPs/NAT): ```python def get_user_id(request: Request) -> str: # Extract from your auth system; fall back to IP user = getattr(request.state, "user", None) return str(user.id) if user else get_remote_address(request) limiter = Limiter(key_func=get_user_id) ``` ## Production: Use Redis as a Backend By default SlowAPI stores counters **in memory**, which breaks if you run multiple workers/instances. Use Redis for shared state: ```bash pip install "slowapi[redis]" # or: pip install redis ``` ```python limiter = Limiter( key_func=get_remote_address, storage_uri="redis://localhost:6379", ) ``` ## Global Default Limit ```python limiter = Limiter( key_func=get_remote_address, default_limits=["200/day", "50/hour"], ) ``` --- ### Notes & Gotchas - `get_remote_address` uses `request.client.host`. If you're behind a proxy/load balancer, you'll need to trust the `X-Forwarded-For` header (e.g., via Starlette's `ProxyHeadersMiddleware` or `uvicorn --proxy-headers`), otherwise all requests appear to come from the proxy's IP. - The `429 Too Many Requests` response includes `Retry-After` and `X-RateLimit-*` headers automatically. Want me to tailor this to your setup (authenticated users, multiple workers, specific endpoints)? Let me know how your app is deployed.

Output

# Rate Limiting in FastAPI I'll show you a **production-ready rate limiting solution** with multiple approaches, from simple to robust. ## Installation ```bash pip install fastapi slowapi redis ``` ## Complete Implementation ### `rate_limiter.py` - Core Setup ```python from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from slowapi.middleware import SlowAPIMiddleware from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import time import asyncio from collections import defaultdict from typing import dict # ── Option 1: SlowAPI (Redis-backed, production-ready) ────────────────────── limiter = Limiter( key_func=get_remote_address, # Rate limit by IP default_limits=["100/minute"], # Global default storage_uri="redis://localhost:6379", # Remove for in-memory ) def create_app() -> FastAPI: app = FastAPI() # Register middleware and error handler app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware(SlowAPIMiddleware) return app ``` ### `main.py` - Routes with Rate Limits ```python from fastapi import FastAPI, Request, Depends, HTTPException from fastapi.responses import JSONResponse from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from slowapi.middleware import SlowAPIMiddleware limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware(SlowAPIMiddleware) # ── Different limits per endpoint ──────────────────────────────────────────── @app.get("/public") @limiter.limit("60/minute") # 60 requests per minute async def public_endpoint(request: Request): return {"message": "Public data"} @app.post("/search") @limiter.limit("10/minute") # Stricter: expensive operation async def search(request: Request): return {"results": []} @app.post("/login") @limiter.limit("5/minute") # Very strict: prevent brute force async def login(request: Request): return {"token": "..."} @app.post("/expensive-ai") @limiter.limit("3/hour") # Per hour for costly operations async def ai_endpoint(request: Request): return {"response": "AI result"} # ── Multiple limits (burst + sustained) ────────────────────────────────────── @app.get("/api/data") @limiter.limit("5/second") # Burst limit @limiter.limit("100/minute") # Sustained limit async def data_endpoint(request: Request): return {"data": "..."} ``` ### `custom_rate_limiter.py` — Rate Limit by User (Not Just IP) ```python from fastapi import FastAPI, Request, Depends, HTTPException from fastapi.security import OAuth2PasswordBearer from slowapi import Limiter from slowapi.errors import RateLimitExceeded from slowapi.middleware import SlowAPIMiddleware import jwt oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # ── Custom key: rate limit by authenticated user ID ─────────────────────────── def get_user_id(request: Request) -> str: """Extract user ID from JWT token, fall back to IP.""" auth = request.headers.get("Authorization", "") if auth.startswith("Bearer "): try: token = auth.split(" ")[1] payload = jwt.decode(token, "secret", algorithms=["HS256"]) return f"user:{payload.get('sub', 'unknown')}" except jwt.InvalidTokenError: pass return f"ip:{request.client.host}" # Fallback to IP limiter = Limiter(key_func=get_user_id) app = FastAPI() app.state.limiter = limiter app.add_middle…

Output

# FastAPI Rate Limiting Pick one: ## Option 1: `slowapi` (easiest) ```bash pip install slowapi ``` ```python from fastapi import FastAPI from slowapi import Limiter from slowapi.util import get_remote_address app = FastAPI() limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter @app.get("/api/endpoint") @limiter.limit("5/minute") async def my_endpoint(request: Request): return {"msg": "ok"} ``` 5 requests per minute per IP. Hits limit → 429 status. ## Option 2: Redis (scalable, multi-server) ```bash pip install aioredis ``` ```python from aioredis import Redis from datetime import datetime, timedelta redis = Redis.from_url("redis://localhost") async def rate_limit(key: str, limit: int = 5, window: int = 60): now = datetime.now().timestamp() pipe = redis.pipeline() pipe.zremrangebyscore(key, 0, now - window) pipe.zadd(key, {str(now): now}) pipe.zcard(key) pipe.expire(key, window) results = await pipe.execute() count = results[2] return count <= limit @app.get("/api/endpoint") async def my_endpoint(request: Request): ip = request.client.host allowed = await rate_limit(f"ratelimit:{ip}") if not allowed: raise HTTPException(status_code=429, detail="Too many requests") return {"msg": "ok"} ``` ## Option 3: In-memory (dev only) ```python from collections import defaultdict from datetime import datetime, timedelta import asyncio request_times = defaultdict(list) async def rate_limit(key: str, limit: int = 5, window: int = 60): now = datetime.now() cutoff = now - timedelta(seconds=window) request_times[key] = [t for t in request_times[key] if t > cutoff] if len(request_times[key]) >= limit: return False request_times[key].append(now) return True @app.get("/api/endpoint") async def my_endpoint(request: Request): ip = request.client.host allowed = await rate_limit(ip) if not allowed: raise HTTPException(status_code=429) return {"msg": "ok"} ``` ## Recommendation - **Dev/single server**: slowapi (simple) - **Production/multi-server**: Redis (persistent, distributed) - **Quick test**: in-memory (don't ship to prod) Which fit your setup?

Output

Use `slowapi`. Wrap FastAPI with limiter. Install: ```bash pip install slowapi ``` Setup: ```python from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from fastapi import FastAPI, Request limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @app.get("/spam-me") @limiter.limit("5/minute") async def my_endpoint(request: Request): return {"ok": True} ``` Key points: - `request: Request` param required. Limiter read it for client IP. - `"5/minute"` = 5 req/min per IP. Tune as need. - Exceed → auto `429 Too Many Requests`. IP limit weak behind proxy/load balancer (all share IP). For prod: use auth user ID as `key_func`, or Redis storage for multi-worker: ```python limiter = Limiter(key_func=get_remote_address, storage_uri="redis://localhost:6379") ```

Output

## Rate Limiting — FastAPI **Easiest:** `slowapi` (wraps Limits library, Redis-compatible). **Install:** ```bash pip install slowapi ``` **Setup:** ```python from fastapi import FastAPI, Request from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @app.get("/your-endpoint") @limiter.limit("10/minute") # ← change to taste async def your_endpoint(request: Request): return {"status": "ok"} ``` **Returns 429** when limit hit. Default response: ```json {"error": "Rate limit exceeded: 10 per 1 minute"} ``` --- ## Options | Need | Solution | |------|----------| | Per-user (auth'd) | `key_func=lambda r: r.state.user_id` | | Multiple limits | `@limiter.limit("10/minute;100/hour")` | | Redis backend (multi-instance) | `Limiter(key_func=..., storage_uri="redis://localhost:6379")` | | Custom 429 response | Replace `_rate_limit_exceeded_handler` with own handler | --- **Redis** = prod standard. Without it, limits live in memory — reset on restart, not shared across workers.

Output

# Rate limiting: minimal setup **What you need:** ```python from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_handler) @app.get("/endpoint") @limiter.limit("5/minute") async def my_endpoint(request: Request): return {"ok": True} ``` **Install:** ```bash pip install slowapi ``` That's it. `slowapi` is the standard for FastAPI, already handles the 429 response, and the decorator works with async. **One thing to know:** This limits per IP. Behind a proxy or load balancer, you'll get the proxy's IP for every user—then configure `limiter = Limiter(key_func=lambda r: r.headers["x-forwarded-for"])` or equivalent. If you need per-user limits (auth token, user ID), swap the key function. **Skipped:** Redis backing (overkill for most), multiple limits per endpoint, custom error messages. Add if you hit the limits of in-memory storage or need distributed rate limiting across servers.

Output

Need one thing before I pick the smallest option: **is your app single-process or running multiple workers/instances?** - **Single process** → use [`slowapi`](https://github.com/laurentS/slowapi) (wraps `limits`, built for FastAPI). ~4 lines, in-memory, zero infra. - **Multiple workers/instances** → same `slowapi` but point it at Redis you already have, else the limit resets per worker. Assuming single process, here's the whole change: ```python from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @app.get("/your-endpoint") @limiter.limit("5/minute") def your_endpoint(request: Request): # `request` param is required by slowapi ... ``` `pip install slowapi`. Adjust `"5/minute"` to taste. One catch worth knowing: `get_remote_address` reads the socket peer. **If you're behind a proxy/load balancer, every request looks like one IP** — you'll need to key on `X-Forwarded-For` instead (and only trust it if the proxy sets it). --- Skipped: Redis setup, per-user (vs per-IP) limiting, and custom 429 messages — none asked for, all easy to add later. Risk: confirm the proxy question above or the limit may apply to all users at once.

Output

**What's your stack?** The right answer depends on two things: 1. **Where does your app run?** Single process / single server, or multiple workers / multiple servers? 2. **Do you already have Redis installed?** --- ## Option A — Single process (no Redis, no new deps) Use `slowapi`, which wraps `limits` and works in-memory with zero infrastructure. ```bash pip install slowapi ``` ```python from fastapi import FastAPI, Request from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @app.get("/your-endpoint") @limiter.limit("10/minute") # ← change to taste async def your_endpoint(request: Request): ... ``` Returns `429 Too Many Requests` automatically. --- ## Option B — Multiple workers/servers (Redis required) Same code, just point `slowapi` at Redis so the counter is shared: ```python limiter = Limiter( key_func=get_remote_address, storage_uri="redis://localhost:6379", # ← your Redis URL ) ``` No other changes. --- ## Pick your key `get_remote_address` is fine unless you're behind a proxy — then it sees the proxy IP and rate-limits everyone together. If you have a proxy, use the forwarded header instead: ```python def get_real_ip(request: Request) -> str: return request.headers.get("X-Forwarded-For", request.client.host).split(",")[0].strip() limiter = Limiter(key_func=get_real_ip) ``` --- **Skipped:** exponential backoff headers (`Retry-After`), per-user (authenticated) limits, and a global limit across all routes — none asked for. **Risk:** `X-Forwarded-For` can be spoofed by clients if your proxy doesn't strip it — only trust that header if your proxy sets it authoritatively.

Output

# Reusable Debounce Function Here's a production-ready debounce implementation: ```javascript function debounce(fn, delay) { let timeoutId; return function debounced(...args) { // Clear the previous timeout if it exists clearTimeout(timeoutId); // Set a new timeout timeoutId = setTimeout(() => { fn.apply(this, args); }, delay); }; } ``` ## Usage Examples ### Basic Example ```javascript // Original function function handleSearch(query) { console.log('Searching for:', query); // API call here } // Create debounced version const debouncedSearch = debounce(handleSearch, 300); // Simulate user typing debouncedSearch('j'); // timeout set debouncedSearch('ja'); // previous cleared, new timeout set debouncedSearch('jav'); // previous cleared, new timeout set debouncedSearch('java'); // previous cleared, new timeout set // After 300ms, logs: "Searching for: java" ``` ### With Event Listeners ```javascript const searchInput = document.querySelector('#search'); searchInput.addEventListener('input', (e) => { debouncedSearch(e.target.value); }); ``` ### Window Resize Example ```javascript function handleResize() { console.log('Window resized to:', window.innerWidth); } window.addEventListener('resize', debounce(handleResize, 250)); ``` ## Enhanced Version with Cancel & Flush For more advanced use cases: ```javascript function debounce(fn, delay) { let timeoutId; const debounced = function(...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => { fn.apply(this, args); }, delay); }; // Cancel pending execution debounced.cancel = () => { clearTimeout(timeoutId); }; // Immediately invoke the function debounced.flush = function(...args) { clearTimeout(timeoutId); fn.apply(this, args); }; return debounced; } ``` ### Usage with Cancel & Flush ```javascript const debouncedSearch = debounce(handleSearch, 300); // Cancel pending call debouncedSearch.cancel(); // Immediately execute without waiting debouncedSearch.flush('search term'); ``` ## Key Points | Feature | Details | |---------|---------| | **Preserves `this`** | Uses `apply()` to maintain correct context | | **Spreads arguments** | Uses rest parameters (`...args`) for flexibility | | **Clears timeout** | Each call clears the previous timeout | | **Returns function** | Returns a new debounced function, not void | This pattern is ideal for search inputs, window resizing, form validation, and any scenario where you need to limit function execution frequency.

Output

# Debounce Function Here's a reusable `debounce` implementation in vanilla JavaScript: ## Basic Version ```javascript function debounce(fn, delay) { let timeoutId; return function (...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => { fn.apply(this, args); }, delay); }; } ``` ## How It Works - **`timeoutId`** is kept in closure scope, persisting between calls. - Each call **clears** the previous pending timer, then schedules a new one. - `fn` only runs once the function hasn't been called for `delay` ms. - `fn.apply(this, args)` preserves the original `this` context and arguments. ## Usage Example ```javascript function handleSearch(query) { console.log('Searching for:', query); } const debouncedSearch = debounce(handleSearch, 300); // Rapid calls — only the last one runs after 300ms debouncedSearch('a'); debouncedSearch('ap'); debouncedSearch('app'); // => "Searching for: app" (after 300ms of inactivity) ``` ```javascript // Typical DOM usage const input = document.querySelector('#search'); input.addEventListener('input', debounce((e) => { console.log(e.target.value); }, 500)); ``` --- ## Enhanced Version (with `immediate` and `cancel`) A more feature-complete version supporting leading-edge calls and cancellation: ```javascript function debounce(fn, delay, immediate = false) { let timeoutId; function debounced(...args) { const callNow = immediate && !timeoutId; clearTimeout(timeoutId); timeoutId = setTimeout(() => { timeoutId = null; if (!immediate) fn.apply(this, args); }, delay); if (callNow) fn.apply(this, args); } // Allow manual cancellation of pending calls debounced.cancel = function () { clearTimeout(timeoutId); timeoutId = null; }; return debounced; } ``` ### Enhanced Usage ```javascript // Leading edge: fire immediately, then ignore subsequent calls for `delay` const log = debounce(() => console.log('Clicked!'), 1000, true); button.addEventListener('click', log); // Cancel a pending invocation log.cancel(); ``` | Option | Behavior | |--------|----------| | `immediate = false` (default) | Trailing edge — runs after calls stop | | `immediate = true` | Leading edge — runs on first call, ignores rest during delay | | `.cancel()` | Clears any pending invocation | **Note:** Use a regular `function` (not an arrow function) for the returned `debounced` so that `this` binds correctly when used as an event handler or method.

Output

# Debounce Function in Vanilla JavaScript ## Core Implementation ```javascript /** * Creates a debounced version of a function that delays execution * until after `delay` milliseconds have elapsed since the last call. * * @param {Function} fn - The function to debounce * @param {number} delay - Milliseconds to wait after last call * @returns {Function} - The debounced function */ function debounce(fn, delay) { let timeoutId = null; function debounced(...args) { // Clear any previously scheduled call clearTimeout(timeoutId); // Schedule a new call after `delay` ms timeoutId = setTimeout(() => { fn.apply(this, args); timeoutId = null; }, delay); } // Cancel any pending execution debounced.cancel = function () { clearTimeout(timeoutId); timeoutId = null; }; // Check if a call is pending debounced.isPending = function () { return timeoutId !== null; }; return debounced; } ``` --- ## Extended Version (with `immediate` option) ```javascript /** * @param {Function} fn - The function to debounce * @param {number} delay - Milliseconds to wait after last call * @param {object} opts * @param {boolean} opts.immediate - If true, call fn on the LEADING edge instead */ function debounce(fn, delay, { immediate = false } = {}) { let timeoutId = null; function debounced(...args) { const callNow = immediate && timeoutId === null; // Reset the timer on every call clearTimeout(timeoutId); timeoutId = setTimeout(() => { timeoutId = null; // Trailing edge: call fn after silence (default behaviour) if (!immediate) { fn.apply(this, args); } }, delay); // Leading edge: call fn immediately on first invocation if (callNow) { fn.apply(this, args); } } debounced.cancel = function () { clearTimeout(timeoutId); timeoutId = null; }; debounced.isPending = function () { return timeoutId !== null; }; return debounced; } ``` --- ## Usage Examples ### 1. Search Input (trailing edge – default) ```javascript const handleSearch = debounce((query) => { console.log('Searching for:', query); // fetch(`/api/search?q=${query}`) }, 400); document.querySelector('#search').addEventListener('input', (e) => { handleSearch(e.target.value); }); // Fires 400ms after the user STOPS typing ``` ### 2. Window Resize Handler ```javascript const handleResize = debounce(() => { console.log('Window size:', window.innerWidth, window.innerHeight); }, 250); window.addEventListener('resize', handleResize); ``` ### 3. Leading Edge (immediate) ```javascript const handleButtonClick = debounce( () => console.log('Button clicked!'), 1000, { immediate: true } ); // Fires IMMEDIATELY on first click, then ignores clicks for 1s document.querySelector('#btn').addEventListener('click', handleButtonClick); ``` ### 4. Cancelling a Pending Call ```javascript const saveData = debounce(() => { console.log('Saving...'); }, 2000); saveData(); // scheduled... saveData.isPending(); // true saveData.cancel(); // cancelled! saveData.isPending(); // false ``` --- ## Behaviour Summary ``` Trailing (default) Leading (immediate: true) ───────────────────────── ───────────────────────── calls: ─A──B──C───── calls: ─A──B──C───── delay ↑ ↑ fires: ──C── fires: ─A────────── ``` | Feature | Description | |---|---| | **Trailing edge** | Fires after the last call + delay (default) | | **Leading edge** | Fires immediately, then ignores for delay ms | | **`.cancel()`** | Cancels any scheduled invocation | | **`.isPending()`** | Returns whether a call is waiting | | **`this` context** | Preserved via `fn.apply(this, args)` | | **Arguments** | All arguments forwarded correctly |

Output

```javascript function debounce(fn, delay) { let timeoutId; return function(...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => fn(...args), delay); }; } ``` How work: Each call clear old timer, start new one. Timer fire only if no new call within `delay` ms. Use: ```javascript const handleSearch = debounce((query) => { console.log('Search:', query); }, 300); input.addEventListener('input', (e) => handleSearch(e.target.value)); ``` Type fast → only last query search. Not fire 10 times per keystroke. Add context preserve (for `this`): ```javascript function debounce(fn, delay) { let timeoutId; return function(...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => fn.apply(this, args), delay); }; } ``` Bonus—return pending status: ```javascript function debounce(fn, delay) { let timeoutId; const debounced = function(...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => fn(...args), delay); }; debounced.cancel = () => clearTimeout(timeoutId); return debounced; } ``` Call `handleSearch.cancel()` kill pending call.

Output

```javascript function debounce(fn, delay) { let timer; return function (...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); }; } ``` How work: each call clear old timer, set new. Only last call survive delay → fn fire once. Use: ```javascript const search = debounce((q) => console.log(q), 300); input.addEventListener("input", (e) => search(e.target.value)); ``` `apply(this, args)` keep caller context + args. Arrow in setTimeout preserve `this`.

Output

```javascript function debounce(fn, delay) { let timer; return function (...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); }; } ``` **How work:** - `timer` close over each debounced fn — no global leak - `clearTimeout` on every call → only last call survive - `fn.apply(this, args)` preserve caller context + all args - Arrow fn in `setTimeout` inherit `this` from wrapper — correct binding **Use:** ```javascript const onResize = debounce(() => recalcLayout(), 200); window.addEventListener("resize", onResize); // Cancel pending call: // clearTimeout exposed if needed — wrap in object for that ``` **Need cancel()?** ```javascript function debounce(fn, delay) { let timer; function debounced(...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); } debounced.cancel = () => clearTimeout(timer); return debounced; } onResize.cancel(); // kill pending call ``` **Edge cases handled:** | Case | Behavior | |------|----------| | Rapid fire calls | Only last executes | | `this` context | Preserved via `apply` | | Arg passing | Full spread, no loss | | Multiple instances | Each own `timer` closure |

Output

```javascript function debounce(fn, delay) { let timeoutId; return function(...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => fn.apply(this, args), delay); }; } ``` **Usage:** ```javascript const search = debounce((query) => console.log('Searching:', query), 300); search('a'); search('ab'); search('abc'); // Only this fires, 300ms later ``` **What I skipped:** leading/trailing edge options, result return, cancellation method. Add only if callers need them—most don't.

Output

```js function debounce(fn, delay) { let timer; return function (...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); }; } ``` Self-check: ```js let n = 0; const inc = debounce(() => n++, 50); inc(); inc(); inc(); setTimeout(() => console.assert(n === 1, `expected 1, got ${n}`), 100); ``` Skipped: leading-edge, cancel/flush methods — add if you need them. `this` is preserved via `apply`, so it works on object methods too.

Output

```js function debounce(fn, delay) { let timer; return function (...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); }; } ``` **Self-check** ```js let calls = 0; const inc = debounce(() => calls++, 50); inc(); inc(); inc(); setTimeout(() => console.assert(calls === 1, 'expected 1 call'), 100); ``` --- Skipped: leading-edge (immediate) option, cancel/flush methods — add when needed. Risk: `timer` is shared across all callers of the returned function, which is the standard behaviour; if you reuse the same debounced reference from two places they share the delay window.

Output

# React Countdown Timer Component Here's a complete countdown timer component with multiple variations: ## Basic Countdown Timer ```jsx import { useState, useEffect } from 'react'; export function CountdownTimer({ initialSeconds = 60 }) { const [seconds, setSeconds] = useState(initialSeconds); const [isActive, setIsActive] = useState(false); useEffect(() => { let interval = null; if (isActive && seconds > 0) { interval = setInterval(() => { setSeconds(seconds => seconds - 1); }, 1000); } else if (seconds === 0) { setIsActive(false); } return () => clearInterval(interval); }, [isActive, seconds]); const toggle = () => setIsActive(!isActive); const reset = () => { setSeconds(initialSeconds); setIsActive(false); }; return ( <div style={styles.container}> <div style={styles.display}>{seconds}</div> <div style={styles.buttonGroup}> <button onClick={toggle} style={styles.button}> {isActive ? 'Pause' : 'Start'} </button> <button onClick={reset} style={styles.button}> Reset </button> </div> </div> ); } const styles = { container: { display: 'flex', flexDirection: 'column', alignItems: 'center', gap: '20px', padding: '20px', }, display: { fontSize: '48px', fontWeight: 'bold', minWidth: '150px', textAlign: 'center', }, buttonGroup: { display: 'flex', gap: '10px', }, button: { padding: '10px 20px', fontSize: '16px', cursor: 'pointer', backgroundColor: '#007bff', color: 'white', border: 'none', borderRadius: '4px', }, }; ``` ## Enhanced Version with Formatting ```jsx import { useState, useEffect } from 'react'; export function FormattedCountdownTimer({ initialSeconds = 3661 }) { const [remainingSeconds, setRemainingSeconds] = useState(initialSeconds); const [isActive, setIsActive] = useState(false); useEffect(() => { let interval = null; if (isActive && remainingSeconds > 0) { interval = setInterval(() => { setRemainingSeconds(prev => prev - 1); }, 1000); } else if (remainingSeconds === 0) { setIsActive(false); } return () => clearInterval(interval); }, [isActive, remainingSeconds]); const formatTime = (totalSeconds) => { const hours = Math.floor(totalSeconds / 3600); const minutes = Math.floor((totalSeconds % 3600) / 60); const seconds = totalSeconds % 60; const pad = (num) => String(num).padStart(2, '0'); if (hours > 0) { return `${pad(hours)}:${pad(minutes)}:${pad(seconds)}`; } return `${pad(minutes)}:${pad(seconds)}`; }; const toggle = () => setIsActive(!isActive); const reset = () => { setRemainingSeconds(initialSeconds); setIsActive(false); }; const isWarning = remainingSeconds <= 10 && remainingSeconds > 0; const isDanger = remainingSeconds <= 5 && remainingSeconds > 0; return ( <div style={styles.container}> <div style={{ ...styles.display, color: isDanger ? '#dc3545' : isWarning ? '#ff9800' : '#333', }} > {formatTime(remainingSeconds)} </div> <div style={styles.buttonGroup}> <button onClick={toggle} style={styles.button}> {isActive ? '⏸ Pause' : '▶ Start'} </button> <button onClick={reset} style={styles.button}> ↻ Reset </button> </div> </div> ); } const styles = { container: { display: 'flex', flexDirection: 'column', alignItems: 'center', gap: '20px', padding: '40px', backgroundColor: '#f5f5f5', borderRadius: '8px', fontFamily: 'monospace', }, display: { fontSize: '64px', fontWeight: 'bold', transition: 'color 0.3s ease', }, buttonGroup: { display: 'flex', gap: '10px', }, button: { padding: '12px 24px', fontSize: '16px', cursor: 'pointer', backgroundColor: '#007bff', color: 'white', border: 'none', borderRadius: '4px', …

Output

# Countdown Timer Component Here's a flexible countdown timer component in React with start, pause, and reset functionality. ```jsx import { useState, useEffect, useRef, useCallback } from 'react'; function CountdownTimer({ initialSeconds = 60, onComplete }) { const [secondsLeft, setSecondsLeft] = useState(initialSeconds); const [isRunning, setIsRunning] = useState(false); const intervalRef = useRef(null); // Clear interval helper const clearTimer = useCallback(() => { if (intervalRef.current) { clearInterval(intervalRef.current); intervalRef.current = null; } }, []); useEffect(() => { if (isRunning && secondsLeft > 0) { intervalRef.current = setInterval(() => { setSecondsLeft((prev) => prev - 1); }, 1000); } return clearTimer; }, [isRunning, secondsLeft, clearTimer]); // Handle completion useEffect(() => { if (secondsLeft === 0 && isRunning) { setIsRunning(false); onComplete?.(); } }, [secondsLeft, isRunning, onComplete]); const handleStart = () => { if (secondsLeft > 0) setIsRunning(true); }; const handlePause = () => setIsRunning(false); const handleReset = () => { setIsRunning(false); setSecondsLeft(initialSeconds); }; // Format as MM:SS const formatTime = (totalSeconds) => { const minutes = Math.floor(totalSeconds / 60); const seconds = totalSeconds % 60; return `${String(minutes).padStart(2, '0')}:${String(seconds).padStart(2, '0')}`; }; return ( <div className="countdown-timer"> <div className="time-display">{formatTime(secondsLeft)}</div> <div className="controls"> {!isRunning ? ( <button onClick={handleStart} disabled={secondsLeft === 0}> Start </button> ) : ( <button onClick={handlePause}>Pause</button> )} <button onClick={handleReset}>Reset</button> </div> </div> ); } export default CountdownTimer; ``` ## Usage ```jsx function App() { return ( <CountdownTimer initialSeconds={120} onComplete={() => alert('Time is up!')} /> ); } ``` ## Optional Styling ```css .countdown-timer { text-align: center; font-family: sans-serif; } .time-display { font-size: 3rem; font-variant-numeric: tabular-nums; margin-bottom: 1rem; } .controls button { margin: 0 0.25rem; padding: 0.5rem 1rem; font-size: 1rem; cursor: pointer; border: none; border-radius: 6px; background: #4f46e5; color: white; } .controls button:disabled { opacity: 0.5; cursor: not-allowed; } ``` ## Key Features - **Start / Pause / Reset** controls - **MM:SS formatting** with `tabular-nums` for stable width - **`onComplete` callback** fires when the timer hits zero - **Cleanup** via `useRef` + `clearInterval` to prevent memory leaks - **Auto-stops** at zero and disables the start button Want me to add anything else — like a progress bar, custom input for setting the time, or sound/notification on completion?

Output

# Countdown Timer Component ## Overview A fully-featured countdown timer built with React that includes start, pause, reset controls, and visual feedback. ## Component Files ### `CountdownTimer.jsx` ```jsx import { useState, useEffect, useRef, useCallback } from "react"; import TimerDisplay from "./TimerDisplay"; import TimerControls from "./TimerControls"; import TimerInput from "./TimerInput"; import styles from "./CountdownTimer.module.css"; const CountdownTimer = ({ initialSeconds = 60 }) => { const [totalSeconds, setTotalSeconds] = useState(initialSeconds); const [timeLeft, setTimeLeft] = useState(initialSeconds); const [isRunning, setIsRunning] = useState(false); const [isFinished, setIsFinished] = useState(false); const intervalRef = useRef(null); const clearTimer = useCallback(() => { if (intervalRef.current) { clearInterval(intervalRef.current); intervalRef.current = null; } }, []); useEffect(() => { if (isRunning && timeLeft > 0) { intervalRef.current = setInterval(() => { setTimeLeft((prev) => { if (prev <= 1) { clearTimer(); setIsRunning(false); setIsFinished(true); return 0; } return prev - 1; }); }, 1000); } return () => clearTimer(); }, [isRunning, timeLeft, clearTimer]); const handleStart = () => { if (timeLeft > 0) { setIsRunning(true); setIsFinished(false); } }; const handlePause = () => { setIsRunning(false); clearTimer(); }; const handleReset = () => { clearTimer(); setIsRunning(false); setIsFinished(false); setTimeLeft(totalSeconds); }; const handleSetTime = (seconds) => { clearTimer(); setIsRunning(false); setIsFinished(false); setTotalSeconds(seconds); setTimeLeft(seconds); }; const progress = totalSeconds > 0 ? (timeLeft / totalSeconds) * 100 : 0; return ( <div className={`${styles.container} ${isFinished ? styles.finished : ""}`}> <h1 className={styles.title}>Countdown Timer</h1> <TimerDisplay timeLeft={timeLeft} progress={progress} isFinished={isFinished} /> <TimerControls isRunning={isRunning} isFinished={isFinished} timeLeft={timeLeft} onStart={handleStart} onPause={handlePause} onReset={handleReset} /> <TimerInput onSetTime={handleSetTime} /> {isFinished && ( <p className={styles.finishedMessage}>⏰ Time&apos;s up!</p> )} </div> ); }; export default CountdownTimer; ``` --- ### `TimerDisplay.jsx` ```jsx import styles from "./CountdownTimer.module.css"; const formatTime = (seconds) => { const hours = Math.floor(seconds / 3600); const minutes = Math.floor((seconds % 3600) / 60); const secs = seconds % 60; if (hours > 0) { return `${String(hours).padStart(2, "0")}:${String(minutes).padStart(2, "0")}:${String(secs).padStart(2, "0")}`; } return `${String(minutes).padStart(2, "0")}:${String(secs).padStart(2, "0")}`; }; const RADIUS = 90; const CIRCUMFERENCE = 2 * Math.PI * RADIUS; const TimerDisplay = ({ timeLeft, progress, isFinished }) => { const strokeDashoffset = CIRCUMFERENCE - (progress / 100) * CIRCUMFERENCE; const strokeColor = isFinished ? "#ef4444" : progress > 50 ? "#22c55e" : progress > 25 ? "#f59e0b" : "#ef4444"; return ( <div className={styles.displayWrapper}> <svg className={styles.progressRing} viewBox="0 0 200 200"> {/* Background circle */} <circle cx="100" cy="100" r={RADIUS} fill="none" stroke="#e5e7eb" strokeWidth="10" /> {/* Progress circle */} <circle cx="100" cy="100" r={RADIUS} fill="none" stroke={strokeColor} strokeWidth="10" strokeLinecap="round" strokeDasharray={CIRCUMFERENCE} strokeDashoffset={strokeDashoffset} …

Output

```jsx import { useState, useEffect } from 'react'; export function CountdownTimer({ initialSeconds = 60 }) { const [seconds, setSeconds] = useState(initialSeconds); const [isActive, setIsActive] = useState(false); useEffect(() => { if (!isActive || seconds <= 0) return; const interval = setInterval(() => { setSeconds(s => s - 1); }, 1000); return () => clearInterval(interval); }, [isActive, seconds]); const toggle = () => setIsActive(!isActive); const reset = () => { setIsActive(false); setSeconds(initialSeconds); }; const minutes = Math.floor(seconds / 60); const secs = seconds % 60; const display = `${minutes}:${secs.toString().padStart(2, '0')}`; return ( <div style={{ textAlign: 'center', padding: '20px' }}> <div style={{ fontSize: '48px', fontWeight: 'bold', marginBottom: '20px' }}> {display} </div> <button onClick={toggle} style={{ marginRight: '10px', padding: '10px 20px' }}> {isActive ? 'Pause' : 'Start'} </button> <button onClick={reset} style={{ padding: '10px 20px' }}> Reset </button> </div> ); } ``` **How it work:** - State: `seconds` (current count), `isActive` (running?) - Effect: run interval only if `isActive` true. Stop at 0. Cleanup interval on unmount/toggle - Display: format as `M:SS` with `padStart` - Buttons: toggle play/pause, reset to initial **Use:** ```jsx <CountdownTimer initialSeconds={120} /> ``` Change `initialSeconds` prop to any number. Timer stop auto at 0, or manual via pause.

Output

```jsx import { useState, useEffect, useRef, useCallback } from 'react'; function CountdownTimer({ seconds = 60, onComplete }) { const [remaining, setRemaining] = useState(seconds); const [running, setRunning] = useState(false); const intervalRef = useRef(null); useEffect(() => { if (!running) return; intervalRef.current = setInterval(() => { setRemaining((prev) => { if (prev <= 1) { clearInterval(intervalRef.current); setRunning(false); onComplete?.(); return 0; } return prev - 1; }); }, 1000); return () => clearInterval(intervalRef.current); }, [running, onComplete]); const start = useCallback(() => { if (remaining > 0) setRunning(true); }, [remaining]); const pause = useCallback(() => setRunning(false), []); const reset = useCallback(() => { setRunning(false); setRemaining(seconds); }, [seconds]); const mins = String(Math.floor(remaining / 60)).padStart(2, '0'); const secs = String(remaining % 60).padStart(2, '0'); return ( <div> <span>{mins}:{secs}</span> <button onClick={start} disabled={running || remaining === 0}>Start</button> <button onClick={pause} disabled={!running}>Pause</button> <button onClick={reset}>Reset</button> </div> ); } export default CountdownTimer; ``` Notes caveman: - `setRemaining` use updater fn → no stale closure. - `onComplete?.()` fire at zero. - Ref hold interval id → clean on unmount + pause. - `seconds` prop drive reset. Usage: `<CountdownTimer seconds={120} onComplete={() => alert('done')} />`

Output

```tsx import { useState, useEffect, useRef } from "react"; interface CountdownTimerProps { initialSeconds: number; } export default function CountdownTimer({ initialSeconds }: CountdownTimerProps) { const [secondsLeft, setSecondsLeft] = useState(initialSeconds); const [running, setRunning] = useState(false); const intervalRef = useRef<ReturnType<typeof setInterval> | null>(null); useEffect(() => { if (running) { intervalRef.current = setInterval(() => { setSecondsLeft((prev) => { if (prev <= 1) { clearInterval(intervalRef.current!); setRunning(false); return 0; } return prev - 1; }); }, 1000); } return () => clearInterval(intervalRef.current!); }, [running]); const reset = () => { clearInterval(intervalRef.current!); setRunning(false); setSecondsLeft(initialSeconds); }; const minutes = Math.floor(secondsLeft / 60); const seconds = secondsLeft % 60; const display = `${String(minutes).padStart(2, "0")}:${String(seconds).padStart(2, "0")}`; const done = secondsLeft === 0; return ( <div style={{ textAlign: "center", fontFamily: "monospace" }}> <div style={{ fontSize: "4rem" }}>{display}</div> {done && <p>Time's up!</p>} <button onClick={() => setRunning(true)} disabled={running || done}> Start </button> <button onClick={() => setRunning(false)} disabled={!running}> Pause </button> <button onClick={reset}>Reset</button> </div> ); } ``` Key parts: **`useEffect` deps on `running`** — interval start/stop tied to state, not manual toggle logic. Clean teardown on re-run via return fn. **Updater fn in `setSecondsLeft`** — `prev => prev - 1` avoids stale closure over `secondsLeft`. Stop logic lives inside updater so interval doesn't need to "know" current value from outer scope. **`intervalRef`** — ref survives re-renders without triggering them. Lets `reset` clear interval from outside the effect. Usage: ```tsx <CountdownTimer initialSeconds={90} /> ```

Output

**Lite version** (what you're asking for): ```jsx export function CountdownTimer({ seconds }) { const [remaining, setRemaining] = React.useState(seconds); React.useEffect(() => { if (remaining <= 0) return; const timer = setInterval(() => setRemaining(r => r - 1), 1000); return () => clearInterval(timer); }, [remaining]); return <div>{remaining}</div>; } ``` **That's it.** Pass in `seconds={10}` and it counts down to zero, then stops. One effect, one state, done. **What I skipped:** pause/resume buttons, format options (MM:SS), restart logic, sound/callbacks when done, animations, accessibility labels. The smallest timer counts and displays. **Risk:** If you need to restart it or change the starting value mid-countdown, pass a key or add a reset function. If you start many timers, each interval stays alive until the component unmounts—fine for a few, watch it if there are dozens.

Output

```jsx import { useState, useEffect } from 'react'; export function CountdownTimer({ seconds }) { const [remaining, setRemaining] = useState(seconds); useEffect(() => { setRemaining(seconds); if (seconds <= 0) return; const id = setInterval(() => { setRemaining((r) => (r <= 1 ? (clearInterval(id), 0) : r - 1)); }, 1000); return () => clearInterval(id); }, [seconds]); return <span>{remaining}</span>; } ``` Usage: `<CountdownTimer seconds={60} />` Skipped (nobody asked): start/pause/reset controls, mm:ss formatting, onComplete callback, custom styling — add if you need them. Risk: `setInterval` drifts over long runs; for minute-plus timers that must stay accurate against the wall clock, switch to comparing `Date.now()` against a target time.

Output

Here's the smallest version that does the job: ```jsx import { useState, useEffect } from "react"; export default function Countdown({ seconds }) { const [left, setLeft] = useState(seconds); useEffect(() => { if (left <= 0) return; const id = setInterval(() => setLeft((n) => n - 1), 1000); return () => clearInterval(id); }, [left]); return <div>{left}s</div>; } ``` **Usage:** ```jsx <Countdown seconds={60} /> ``` When `left` hits 0 the interval stops. The effect re-runs on each tick, so the interval is always fresh — no stale-closure bugs. --- **Skipped:** pause/reset controls, formatting as MM:SS, onComplete callback, styling — none were asked for. **Risk:** `seconds` prop changes after mount are ignored; if you need that, reset state inside an effect that watches `seconds`.

Output

# Python code to read sales.csv and sum the 'amount' column ```python import pandas as pd # Read the CSV file df = pd.read_csv('sales.csv') # Sum the 'amount' column total_amount = df['amount'].sum() # Print the result print(f"Total Amount: {total_amount}") ``` ## Alternative methods: **Using built-in CSV module:** ```python import csv total = 0 with open('sales.csv', 'r') as file: reader = csv.DictReader(file) for row in reader: total += float(row['amount']) print(f"Total Amount: {total}") ``` **Using NumPy:** ```python import numpy as np import pandas as pd df = pd.read_csv('sales.csv') total_amount = np.sum(df['amount'].values) print(f"Total Amount: {total_amount}") ``` ## Notes: - **Pandas** (first method) is the most concise and recommended for data analysis - The **CSV module** approach is good if you want to avoid external dependencies - Make sure `sales.csv` is in the same directory as your script, or provide the full path - The 'amount' column values should be numeric (int or float)

Output

# Sum the 'amount' Column in sales.csv Here are a few ways to accomplish this, depending on your needs. ## Option 1: Using the `csv` module (standard library, no dependencies) ```python import csv total = 0.0 with open('sales.csv', newline='', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: # Skip empty or missing values gracefully value = row.get('amount', '').strip() if value: total += float(value) print(f"Total amount: {total:.2f}") ``` ## Option 2: Using pandas (cleaner for data work) ```python import pandas as pd df = pd.read_csv('sales.csv') total = df['amount'].sum() print(f"Total amount: {total:.2f}") ``` ## Option 3: Robust version with error handling ```python import csv def sum_amount(filename, column='amount'): total = 0.0 skipped = 0 try: with open(filename, newline='', encoding='utf-8') as f: reader = csv.DictReader(f) if column not in reader.fieldnames: raise KeyError(f"Column '{column}' not found. " f"Available: {reader.fieldnames}") for line_num, row in enumerate(reader, start=2): value = row.get(column, '').strip() try: total += float(value) except (ValueError, TypeError): skipped += 1 # non-numeric or empty cell except FileNotFoundError: print(f"Error: '{filename}' not found.") return None if skipped: print(f"Warning: skipped {skipped} non-numeric row(s).") return total result = sum_amount('sales.csv') if result is not None: print(f"Total amount: {result:.2f}") ``` ## Notes - **Option 1** is best if you can't install packages. - **Option 2** is simplest if you already use pandas (install with `pip install pandas`). - **Option 3** handles missing files, missing columns, and bad data. If your CSV uses a different delimiter (e.g., `;`), pass `delimiter=';'` to `csv.DictReader` or `sep=';'` to `pd.read_csv`. Let me know your column names or a sample of the file if you'd like this tailored further.

Output

## Reading and Summing Sales Data Here's a clean Python solution with error handling: ```python import csv import os def sum_sales_amount(filepath: str = "sales.csv") -> float: """ Reads a CSV file and sums the 'amount' column. Args: filepath: Path to the CSV file (default: 'sales.csv') Returns: Total sum of the 'amount' column as a float Raises: FileNotFoundError: If the CSV file doesn't exist KeyError: If the 'amount' column is missing ValueError: If any 'amount' value cannot be converted to a number """ if not os.path.exists(filepath): raise FileNotFoundError(f"File not found: '{filepath}'") total = 0.0 row_count = 0 errors = [] with open(filepath, newline="", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile) # Validate that 'amount' column exists if reader.fieldnames is None or "amount" not in reader.fieldnames: raise KeyError("Column 'amount' not found in CSV file.") for line_num, row in enumerate(reader, start=2): # start=2 (header is line 1) raw_value = row["amount"].strip() try: total += float(raw_value) row_count += 1 except ValueError: errors.append(f" Line {line_num}: invalid value '{raw_value}'") # Report any skipped rows if errors: print(f"⚠️ Skipped {len(errors)} invalid row(s):") for error in errors: print(error) print(f"✅ Processed {row_count} valid row(s).") print(f"💰 Total sales amount: {total:,.2f}") return total if __name__ == "__main__": total = sum_sales_amount("sales.csv") ``` --- ### Example `sales.csv` ``` date,product,amount 2024-01-01,Widget A,150.00 2024-01-02,Widget B,200.50 2024-01-03,Widget C,99.99 2024-01-04,Widget D,350.00 ``` ### Example Output ``` ✅ Processed 4 valid row(s). 💰 Total sales amount: 800.49 ``` --- ### Alternative: Using `pandas` (simpler for large files) ```python import pandas as pd def sum_sales_amount_pandas(filepath: str = "sales.csv") -> float: """Reads a CSV and sums the 'amount' column using pandas.""" df = pd.read_csv(filepath) if "amount" not in df.columns: raise KeyError("Column 'amount' not found in CSV file.") total = pd.to_numeric(df["amount"], errors="coerce").sum() print(f"✅ Processed {len(df)} row(s) ({df['amount'].isna().sum()} skipped).") print(f"💰 Total sales amount: {total:,.2f}") return total if __name__ == "__main__": total = sum_sales_amount_pandas("sales.csv") ``` --- ### Key Features | Feature | Description | |---|---| | **Error handling** | Catches missing file, missing column, and bad values | | **Row tracking** | Reports how many rows were processed or skipped | | **Flexible input** | Accepts any file path as an argument | | **Two approaches** | Built-in `csv` module or `pandas` for large datasets | Use the **`csv` version** for no dependencies, or the **`pandas` version** for larger/more complex datasets.

Output

```python import pandas as pd df = pd.read_csv('sales.csv') total = df['amount'].sum() print(total) ``` Done. Reads CSV, sums `amount` column, prints result. Need grouping, filtering, or file check first?

Output

```python import csv with open("sales.csv") as f: total = sum(float(row["amount"]) for row in csv.DictReader(f)) print(total) ``` Pandas way if big file: ```python import pandas as pd print(pd.read_csv("sales.csv")["amount"].sum()) ```

Output

```python import pandas as pd df = pd.read_csv('sales.csv') total = df['amount'].sum() print(f"Total sales: {total}") ``` No pandas? Pure stdlib: ```python import csv total = 0.0 with open('sales.csv', newline='') as f: for row in csv.DictReader(f): total += float(row['amount']) print(f"Total sales: {total}") ```

Output

```python import pandas as pd df = pd.read_csv('sales.csv') total = df['amount'].sum() print(total) ``` That's it. If `sales.csv` doesn't exist or `amount` is missing, it fails loud—which is right for a script. **Skipped:** error handling (file not found, bad data), parsing amount as numeric (pandas does it), output format. **Risk:** if amount has non-numeric values, `.sum()` skips them silently; add `.astype(float)` before `.sum()` if that matters.

Output

```python import csv with open("sales.csv", newline="") as f: total = sum(float(row["amount"]) for row in csv.DictReader(f)) print(total) ``` Skipped: error handling for missing file/column/bad values, and I didn't see your CSV so I'm assuming a header row with an `amount` column. Risk: crashes on an empty or non-numeric `amount` cell.

Output

```python import csv with open("sales.csv") as f: print(sum(float(row["amount"]) for row in csv.DictReader(f))) ``` --- Skipped: error handling for missing file or bad values. Risk: crashes on empty `amount` cells or non-numeric data.

Output

# Email Validation Function Here are a few approaches, from simple to more comprehensive: ## 1. **Simple Regex Approach** (Quick & Easy) ```python import re def validate_email_simple(email: str) -> bool: """ Simple email validation using regex. """ pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None # Usage print(validate_email_simple("user@example.com")) # True print(validate_email_simple("invalid.email@")) # False ``` ## 2. **RFC 5322 Compliant** (More Thorough) ```python import re def validate_email_rfc5322(email: str) -> bool: """ More comprehensive validation following RFC 5322 standards. """ pattern = r'^[a-zA-Z0-9.!#$%&\'*+/=?^_`{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$' return re.match(pattern, email) is not None # Usage print(validate_email_rfc5322("user+tag@example.co.uk")) # True print(validate_email_rfc5322("user@localhost")) # True ``` ## 3. **Using `email-validator` Library** (Recommended for Production) ```python # Install: pip install email-validator from email_validator import validate_email, EmailNotValidError def validate_email_professional(email: str) -> tuple[bool, str]: """ Professional validation with error messages. Returns (is_valid, message) """ try: valid_email = validate_email(email) return True, f"Valid email: {valid_email.normalized}" except EmailNotValidError as e: return False, str(e) # Usage is_valid, message = validate_email_professional("user@example.com") print(is_valid, message) # True, Valid email: user@example.com is_valid, message = validate_email_professional("invalid@") print(is_valid, message) # False, The email address is not valid... ``` ## 4. **Complete Class-Based Solution** ```python import re from typing import Optional class EmailValidator: """Comprehensive email validator with multiple validation levels.""" SIMPLE_PATTERN = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' RFC5322_PATTERN = r'^[a-zA-Z0-9.!#$%&\'*+/=?^_`{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$' @staticmethod def validate(email: str, strict: bool = False) -> bool: """ Validate email address. Args: email: Email address to validate strict: If True, use RFC5322 pattern; otherwise use simple pattern Returns: True if valid, False otherwise """ if not isinstance(email, str) or not email.strip(): return False email = email.strip().lower() pattern = EmailValidator.RFC5322_PATTERN if strict else EmailValidator.SIMPLE_PATTERN return re.match(pattern, email) is not None @staticmethod def validate_with_details(email: str) -> dict: """Validate and return detailed information.""" return { 'email': email, 'is_valid_simple': EmailValidator.validate(email, strict=False), 'is_valid_strict': EmailValidator.validate(email, strict=True), 'length': len(email), 'has_plus': '+' in email, 'domain': email.split('@')[1] if '@' in email else None } # Usage print(EmailValidator.validate("user@example.com")) # True print(EmailValidator.validate("user+tag@example.com", strict=True)) # True details = EmailValidator.validate_with_details("test.user@sub.example.co.uk") print(details) ``` ## Comparison Table | Method | Pros | Cons | |--------|------|------| | **Simple Regex** | Fast, no dependencies | Limited validation | | **RFC 5322** | More comprehensive | Complex, false positives possible | | **email-validator** | Most accurate, checks DNS | Requires dependency, slower | | **Class-Based** | Flexible, reusable | More code | ## Recommendation - **Quick project**: Use method #1 - **Production app**: Use method #3 (`email-validator`) - **Ne…

Output

# Email Validation in Python Here are a few approaches, from simple to robust. ## Option 1: Using Regex (Practical Approach) ```python import re def is_valid_email(email): """ Validate an email address using a regex pattern. Returns True if valid, False otherwise. """ if not isinstance(email, str): return False pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None ``` ## Option 2: More Thorough Validation ```python import re def validate_email(email): """ Validate an email with additional checks. Returns (is_valid: bool, message: str). """ if not isinstance(email, str): return False, "Email must be a string" email = email.strip() if not email: return False, "Email cannot be empty" if email.count('@') != 1: return False, "Email must contain exactly one '@'" local, domain = email.split('@') if not local: return False, "Missing local part (before '@')" if len(local) > 64: return False, "Local part too long (max 64 chars)" if len(email) > 254: return False, "Email too long (max 254 chars)" pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' if not re.match(pattern, email): return False, "Invalid email format" return True, "Valid email" ``` ## Option 3: Using a Library (Recommended for Production) For real-world applications, use the [`email-validator`](https://pypi.org/project/email-validator/) library, which handles edge cases and can even verify deliverability. ```bash pip install email-validator ``` ```python from email_validator import validate_email, EmailNotValidError def check_email(email): try: # check_deliverability=False skips DNS lookup validated = validate_email(email, check_deliverability=False) return True, validated.normalized except EmailNotValidError as e: return False, str(e) ``` ## Usage Examples ```python emails = [ "user@example.com", # valid "john.doe@company.co.uk",# valid "invalid.email", # no @ "@example.com", # no local part "user@@example.com", # double @ ] for email in emails: print(f"{email:25} -> {is_valid_email(email)}") ``` ## Important Notes - **Regex has limits:** The official email spec ([RFC 5322](https://datatracker.ietf.org/doc/html/rfc5322)) is extremely complex. No simple regex catches every valid/invalid case perfectly. - **The only true test** of whether an email exists is to send a confirmation message to it. - For most apps, **Option 1** is fine; for production, prefer **Option 3**. Let me know your use case and I can tailor the solution!

Output

```python import re def validate_email(email: str) -> bool: """ Validate email format. Args: email: Email string to validate Returns: True if valid, False otherwise """ pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None ``` Use: `validate_email("user@example.com")` → `True` Pattern check: - `^` start - `[a-zA-Z0-9._%+-]+` local part (letters, numbers, dots, underscore, %, plus, dash) - `@` separator - `[a-zA-Z0-9.-]+` domain name - `\.` dot - `[a-zA-Z]{2,}` TLD (2+ letters) - `$` end **Note:** RFC 5322 full spec way more complex. This cover ~95% real cases. For strict validation, use `email-validator` package. ```python from email_validator import validate_email, EmailNotValidError try: validate_email("user@example.com") except EmailNotValidError: print("Invalid") ```

Output

```python import re def validate_email(email: str) -> bool: pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" return bool(re.match(pattern, email)) ``` Regex check basic format. Good enough most cases. Edge cases (quoted local, IP domain) need full RFC 5322 parser. For real signup: send confirm email = only true test.

Output

```python import re def validate_email(email: str) -> bool: pattern = r'^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$' return bool(re.match(pattern, email)) ``` Checks: local part chars, `@` present, domain valid, TLD ≥2 chars. Won't catch: disposable domains, MX record existence. For production, use `email-validator` lib.

Output

```python import re _EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") def is_valid_email(email: str) -> bool: return bool(_EMAIL_RE.match(email)) ``` ```python assert is_valid_email("a@b.com") assert not is_valid_email("a@b") assert not is_valid_email("@b.com") assert not is_valid_email("a b@c.com") ``` Skipped: full RFC 5322 parsing and no MX/deliverability check — this only confirms the shape looks like an email. If you actually need to know an address works, send a confirmation email instead; no regex can tell you that.

Output

Here's the smallest thing that actually works for real-world email validation: ```python import re # Covers 99%+ of real addresses; does not parse RFC 5321 edge cases _EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") def is_valid_email(email: str) -> bool: return bool(_EMAIL_RE.match(email)) ``` **Usage** ```python is_valid_email("user@example.com") # True is_valid_email("bad@") # False ``` --- If you need anything stricter, just send a real delivery attempt with [smtplib](https://docs.python.org/3/library/smtplib.html) or use the `email-validator` package. **Skipped:** quoted local parts, IP-address domains, Unicode TLDs, and MX-record checks — all rare enough to ignore unless you have a specific failing case. **Risk:** a creative-but-valid RFC 5321 address (e.g. `"foo bar"@example.com`) will be rejected.