evalship-hq / ponytail

Output

# Reading Temperature from a Thermistor via MCP3008 Here's a complete 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 Raspberry Pi via `sudo raspi-config` → Interface Options → SPI. ## Code ```python import math import time import board import busio import digitalio import adafruit_mcp3xxx.mcp3008 as MCP from adafruit_mcp3xxx.analog_in import AnalogIn def read_thermistor_temp( channel=MCP.P0, vcc=3.3, series_resistor=10000.0, # Fixed resistor in the voltage divider (ohms) nominal_resistance=10000.0, # Thermistor resistance at nominal temp (ohms) nominal_temp_c=25.0, # Nominal temperature (Celsius) beta_coefficient=3950.0, # Beta value from thermistor datasheet ): """ Read temperature in Celsius from a thermistor via MCP3008 channel 0. Wiring (voltage divider): VCC --- [series_resistor] ---+--- [thermistor] --- GND | MCP3008 CH0 Returns: Temperature in degrees Celsius (float). """ # --- Set up SPI bus and MCP3008 --- 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) # Create an analog input on the specified channel chan = AnalogIn(mcp, channel) # --- Read the ADC voltage --- # chan.voltage already gives the measured voltage at the ADC pin v_out = chan.voltage if v_out <= 0: raise ValueError("Measured voltage is zero; check wiring.") # --- Compute thermistor resistance --- # Voltage divider: v_out = vcc * (R_therm / (R_series + R_therm)) # Solve for R_therm: r_therm = series_resistor * (v_out / (vcc - v_out)) # --- Apply the Beta (Steinhart) equation --- # 1/T = 1/T0 + (1/B) * ln(R/R0) # Temperatures in Kelvin t0_kelvin = nominal_temp_c + 273.15 inv_t = (1.0 / t0_kelvin) + (1.0 / beta_coefficient) * math.log( r_therm / nominal_resistance ) temp_kelvin = 1.0 / inv_t temp_celsius = temp_kelvin - 273.15 return temp_celsius if __name__ == "__main__": while True: try: temp = read_thermistor_temp() print(f"Temperature: {temp:.2f} °C") except Exception as e: print(f"Error reading temperature: {e}") time.sleep(1) ``` ## Notes **1. Wiring matters for the math** The code assumes this voltage divider (thermistor on the low/GND side): ``` VCC --- [R_series] ---+--- CH0 | [Thermistor] --- GND ``` If you swap the thermistor and resistor positions, change the resistance calculation to: ```python r_therm = series_resistor * ((vcc - v_out) / v_out) ``` **2. Tune the parameters to your hardware** Grab these from your thermistor's datasheet: - `nominal_resistance` — e.g. 10 kΩ (common NTC) - `nominal_temp_c` — usually 25 °C - `beta_coefficient` — e.g. 3950 (varies by part) **3. Efficiency improvement** For repeated reads, initialize the SPI/MCP objects **once** instead of per-call: ```python class ThermistorReader: def __init__(self, channel=MCP.P0, cs_pin=board.D5, **params): spi = busio.SPI(clock=board.SCK, MISO=board.MISO, MOSI=board.MOSI) cs = digitalio.DigitalInOut(cs_pin) self.mcp = MCP.MCP3008(spi, cs) self.chan = AnalogIn(self.mcp, channel) self.params = params def read_temp(self): v_out = self.chan.voltage r_therm = self.params.get("series_resistor", 10000.0) * ( v_out / (self.params.get("vcc", 3.3) - v_out) ) t0 = self.params.get("nominal_temp_c", 25.0) + 273.15 beta = self.params.get("beta_coefficient", 3950.0) r0 = self.params.get("nominal_resistance", 10000.0) inv_t = (1 / t0) + (1 / beta) * math.log(r_therm / r0) return (1 / inv_t) - 273.15 ``` **4…

Output

```python import spidev import math # Beta-model thermistor on MCP3008 CH0. 3.3V ref, 10kΩ series resistor, 10kΩ NTC @ 25°C, B=3950. # Calibrate R_SERIES, R0, T0, B to your actual parts. VREF = 3.3 R_SERIES = 10000.0 R0 = 10000.0 T0 = 298.15 # 25°C in Kelvin BETA = 3950.0 _spi = spidev.SpiDev() _spi.open(0, 0) _spi.max_speed_hz = 1350000 def read_temperature_c(channel=0): raw = _spi.xfer2([1, (8 + channel) << 4, 0]) adc = ((raw[1] & 3) << 8) | raw[2] # 0..1023 if adc == 0 or adc == 1023: raise ValueError(f"ADC reading {adc} out of range (open/short thermistor?)") voltage = adc / 1023.0 * VREF r_therm = R_SERIES * voltage / (VREF - voltage) # thermistor on low side, series to VREF kelvin = 1.0 / (1.0 / T0 + math.log(r_therm / R0) / BETA) return kelvin - 273.15 ``` Wiring this assumes: 3.3V → R_SERIES → CH0 → thermistor → GND. If yours is flipped, swap the `r_therm` formula to `R_SERIES * (VREF - voltage) / voltage`. Skipped: averaging multiple samples (add later if readings jitter), and the Steinhart-Hart 3-coefficient model (Beta is fine for a ±5°C hobby range). Not checked: your actual B/R0 values — the defaults are placeholders you must calibrate, or temps will be wrong.

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): """ Parse a duration string into total seconds. Supports the following units: - 'd' (days) - 'h' (hours) - 'm' (minutes) - 's' (seconds) Examples: '1h30m45s' -> 5445 '2d' -> 172800 '90m' -> 5400 '45s' -> 45 Args: duration_str (str): The duration string to parse. Returns: int: Total number of seconds. Raises: ValueError: If the string is empty or contains invalid tokens. """ if not duration_str or not duration_str.strip(): raise ValueError("Duration string cannot be empty") duration_str = duration_str.strip().lower() # Mapping of unit characters to their value in seconds unit_seconds = { 'd': 86400, 'h': 3600, 'm': 60, 's': 1, } # Match number + unit pairs, e.g. "1h", "30m", "45s" pattern = re.compile(r'(\d+)([dhms])') matches = pattern.findall(duration_str) # Verify the entire string was consumed (no leftover/invalid characters) consumed = ''.join(num + unit for num, unit in matches) if consumed != duration_str: raise ValueError(f"Invalid duration string: {duration_str!r}") total_seconds = 0 for value, unit in matches: total_seconds += int(value) * unit_seconds[unit] return total_seconds ``` ## 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 # Invalid inputs raise ValueError try: parse_duration('1x30y') except ValueError as e: print(e) # Invalid duration string: '1x30y' ``` ## How It Works 1. **Normalization** — Strips whitespace and lowercases the input. 2. **Regex matching** — `(\d+)([dhms])` captures each number/unit pair. 3. **Validation** — Reconstructs the matched string and compares it to the original to ensure there are no invalid or leftover characters. 4. **Summation** — Multiplies each value by its corresponding factor in seconds. ## Optional: Support for Fractional Values If you want to allow decimal values like `'1.5h'`: ```python def parse_duration_float(duration_str): """Parse a duration string with optional decimal values into total seconds.""" if not duration_str or not duration_str.strip(): raise ValueError("Duration string cannot be empty") duration_str = duration_str.strip().lower() unit_seconds = {'d': 86400, 'h': 3600, 'm': 60, 's': 1} pattern = re.compile(r'(\d+(?:\.\d+)?)([dhms])') matches = pattern.findall(duration_str) consumed = ''.join(num + unit for num, unit in matches) if consumed != duration_str: raise ValueError(f"Invalid duration string: {duration_str!r}") return sum(float(value) * unit_seconds[unit] for value, unit in matches) print(parse_duration_float('1.5h')) # 5400.0 ``` Let me know if you'd like it to support additional units (like weeks or milliseconds) or return a `timedelta` object instead!

Output

**What's your setup?** Before I write anything, I need to know: 1. **How do you identify users?** (API key in header, JWT, session cookie, IP address?) 2. **What limit?** (e.g. 10 req/min per user) 3. **Do you have Redis?** (or is this a single-process app where in-memory is fine?) --- While you answer — the honest shortlist: | Option | When it fits | |---|---| | **`slowapi`** (1 decorator) | Already in your deps, or you want standard `Limits` header support | | **In-memory dict + middleware** | Single process, no Redis, no install | | **Reverse proxy rule** (nginx/Caddy/Cloudflare) | You control infra — zero app code | The reverse-proxy option is free and the most robust. If that's available, it beats adding code entirely. --- **Skipped:** the actual implementation — answering blind would likely give you the wrong key, wrong limit, or an approach that breaks under your auth model. **Risk:** in-memory rate limiting resets on restart and doesn't work across multiple workers/replicas.

Output

## Email Validation Function Here's a Python function that validates email addresses using both regex and logical checks: ```python import re def validate_email(email: str) -> dict: """ Validates an email address and returns a detailed result. Args: email: The email address to validate Returns: A dict with 'is_valid' (bool) and 'message' (str) keys """ # Check if email is a non-empty string if not isinstance(email, str) or not email: return {"is_valid": False, "message": "Email must be a non-empty string"} # Check 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("@") # Validate local part (before @) 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"} # Validate domain part (after @) if not domain: return {"is_valid": False, "message": "Domain part (after '@') cannot be empty"} if "." not in domain: return {"is_valid": False, "message": "Domain must contain at least one '.'"} if domain.startswith(".") or domain.endswith("."): return {"is_valid": False, "message": "Domain cannot start or end with a '.'"} # Full regex pattern for thorough validation pattern = r"^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$" if not re.match(pattern, email): return {"is_valid": False, "message": "Email format is invalid"} return {"is_valid": True, "message": "Email is valid"} def is_valid_email(email: str) -> bool: """Simple wrapper that returns just a boolean.""" return validate_email(email)["is_valid"] ``` ### Usage Examples ```python test_emails = [ "user@example.com", # Valid "user.name+tag@domain.co", # Valid "invalid-email", # No @ symbol "@domain.com", # Empty local part "user@", # Empty domain "user@domain", # No TLD "user@@domain.com", # Multiple @ symbols "a" * 65 + "@domain.com", # Local part too long "user@.domain.com", # Domain starts with dot ] for email in test_emails: result = validate_email(email) status = "✅" if result["is_valid"] else "❌" print(f"{status} '{email}': {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 ❌ '@domain.com': Local part (before '@') cannot be empty ❌ 'user@': Domain part (after '@') cannot be empty ❌ 'user@domain': Domain must contain at least one '.' ❌ 'user@@domain.com': Email must contain exactly one '@' symbol ❌ 'aaaaa...@domain.com': Local part exceeds maximum length of 64 characters ❌ 'user@.domain.com': Domain cannot start or end with a '.' ``` ### Key Features | Feature | Details | |---|---| | **Type check** | Ensures input is a non-empty string | | **Length limits** | 254 chars total, 64 chars for local part (RFC 5321) | | **Structure checks** | Exactly one `@`, valid domain format | | **Regex validation** | Catches remaining format issues | | **Detailed feedback** | Returns specific error messages | > **Note:** For production use, consider a library like **`email-validator`** (`pip install email-validator`) which handles edge cases and checks DNS records for domain existence.

Output

```python import re def parse_duration(s): units = {'h': 3600, 'm': 60, 's': 1} pairs = re.findall(r'(\d+)([hms])', s) if not pairs or re.sub(r'\d+[hms]', '', s): raise ValueError(f"invalid duration: {s!r}") return sum(int(n) * units[u] for n, u in pairs) assert parse_duration('1h30m45s') == 5445 assert parse_duration('45s') == 45 assert parse_duration('2h') == 7200 for bad in ('', 'abc', '1h30x', '30'): try: parse_duration(bad); assert False except ValueError: pass ``` Skipped: days/weeks, decimals (`1.5h`), and order/duplicate checks (`30m1h` and `1h1h` both parse). Add if your inputs need them.

Output

# Refactored Code ```python def double_positive_values(items): """Return a list of doubled 'a' values for items where 'a' is positive. Args: items: An iterable of dictionaries, each potentially containing key 'a'. Returns: A list of integers/floats, each being the 'a' value multiplied by 2. """ doubled_values = [] for item in items: value = item.get('a') if value is not None and value > 0: doubled_values.append(value * 2) return doubled_values ``` Or, if you prefer a more concise, Pythonic version: ```python def double_positive_values(items): """Return a list of doubled 'a' values for items where 'a' is positive.""" return [item['a'] * 2 for item in items if item.get('a') is not None and item['a'] > 0] ``` --- # Step-by-Step Write-Up of Every Change ### 1. Renamed the function `p` → `double_positive_values` **Why:** `p` is meaningless. A reader has no idea what the function does without reading the body. The new name describes the *intent* (double the values) and the *condition* (only positive ones). Good function names act as free documentation. ### 2. Renamed the parameter `d` → `items` **Why:** `d` presumably meant "data" or "dict," but it's actually an iterable *of* dictionaries. `items` communicates that we're iterating over a collection of elements. (If you know the domain, an even more specific name like `records` or `users` would be better.) ### 3. Renamed the result variable `r` → `doubled_values` **Why:** `r` ("result") is generic. `doubled_values` tells the reader exactly what the list contains, making the `return` statement self-explanatory. ### 4. Renamed the loop variable `x` → `item` **Why:** `x` is an anonymous placeholder. `item` makes it clear each element is one record from the collection, which reads naturally alongside `for item in items`. ### 5. Extracted `x['a']` into a `value` variable **Why:** The original code accessed the dictionary twice: once via `x.get('a')` and once via `x['a']`. This is: - **Repetitive** — the key `'a'` is duplicated. - **Slightly inefficient** — two lookups instead of one. - **Potentially inconsistent** — if the dict changed between calls (unlikely here, but a code smell). Storing it once in `value` fixes all three. ### 6. Changed the truthiness check `if x.get('a')` → `if value is not None` **Why — this is the most important correctness fix.** The original `x.get('a')` relies on truthiness, which has a subtle bug: - `get('a')` returns `None` if the key is missing — this is the intended guard. - **But** it also treats `0`, `0.0`, and `False` as "absent," short-circuiting before the `> 0` check. While `0 > 0` is `False` anyway (so behavior happens to match for `0`), the intent is muddled. Using `is not None` **explicitly** separates two distinct concerns: 1. **Does the key exist?** → `value is not None` 2. **Is it positive?** → `value > 0` This makes the logic clear and avoids accidentally relying on truthiness coincidences. > ⚠️ Note: If `'a'` can legitimately hold `None`, the explicit check also prevents a `TypeError` from comparing `None > 0` (which raises in Python 3). The original `get('a')` truthiness check masked this; my version guards it intentionally. ### 7. Added a docstring **Why:** It documents the arguments, return value, and intent so callers don't need to read the implementation. This is standard for any reusable function. --- ### Summary of behavioral difference The two versions behave identically for normal numeric inputs. The refactor **intentionally makes the "missing key" guard explicit** (`is not None`) rather than relying on truthiness. If your data could contain `'a': None`, the refactored version is also **safer** (no `TypeError`) and **clearer** about intent. If you actually *wanted* to skip zero/falsy values, let me know and I'll adjust the condition accordingly.

Output

Lazy read: I don't know what `d`, `a`, or the doubling mean, so names are generic guesses. ```python def doubled_positives(items): return [item['a'] * 2 for item in items if item.get('a', 0) > 0] ``` Changes: 1. `p` → `doubled_positives`, `d` → `items`, `x` → `item`: tells the reader what it does without decoding. 2. Loop + `append` → list comprehension: standard Python for "build a list by filtering and mapping"; removes the `r = []` / `return r` boilerplate. 3. `x.get('a') and x['a'] > 0` → `item.get('a', 0) > 0`: the original double-checks because `.get('a')` can be `None` and `None > 0` crashes. Defaulting to `0` makes the single comparison safe and drops the redundant truthiness check. One behavior note: original `and` also filtered out `0` and `None` via truthiness; the new `> 0` keeps the same result (0 and missing both fail `> 0`), so no change for callers. Skipped: real names (don't know the domain), type hints, and any test — trivial one-liner, and I can't assert without knowing valid inputs. Risk: if any `item['a']` is a non-number (e.g. a string), both versions crash the same way — behavior unchanged.

Output

# Reading and summing the 'amount' column from sales.csv Here are a few different approaches: ## Method 1: Using Pandas (Recommended) ```python import pandas as pd # Read the CSV file df = pd.read_csv('sales.csv') # Sum the 'amount' column total = df['amount'].sum() print(f"Total sales: ${total:.2f}") ``` ## Method 2: Using CSV Module ```python import csv total = 0 # Read the CSV file with open('sales.csv', 'r') as file: reader = csv.DictReader(file) for row in reader: total += float(row['amount']) print(f"Total sales: ${total:.2f}") ``` ## Method 3: Using CSV Module (Simple) ```python import csv total = sum(float(row['amount']) for row in csv.DictReader(open('sales.csv'))) print(f"Total sales: ${total:.2f}") ``` ## Method 4: Using NumPy ```python import numpy as np data = np.genfromtxt('sales.csv', delimiter=',', dtype=str, skip_header=1) total = np.sum(data[:, 1].astype(float)) # Adjust column index as needed print(f"Total sales: ${total:.2f}") ``` ## Recommended Approach **Method 1 (Pandas)** is the best choice because it: - ✅ Handles different data types automatically - ✅ Is readable and maintainable - ✅ Provides error handling for missing values - ✅ Works with large files efficiently ### Complete Example with Error Handling: ```python import pandas as pd try: df = pd.read_csv('sales.csv') total = df['amount'].sum() print(f"Total sales: ${total:.2f}") except FileNotFoundError: print("Error: sales.csv not found") except KeyError: print("Error: 'amount' column not found") ```

Output

# Reading sales.csv and Summing the 'amount' Column Here are a few approaches depending on your needs and available libraries. ## Option 1: Using pandas (recommended for data work) ```python import pandas as pd # Read the CSV file df = pd.read_csv('sales.csv') # Sum the 'amount' column total = df['amount'].sum() print(f"Total amount: {total}") ``` ## Option 2: Using the built-in csv module (no dependencies) ```python import csv total = 0.0 with open('sales.csv', newline='') as f: reader = csv.DictReader(f) for row in reader: total += float(row['amount']) print(f"Total amount: {total}") ``` ## Option 3: Robust version with error handling ```python import csv def sum_amount(filename='sales.csv', column='amount'): total = 0.0 skipped = 0 try: with open(filename, newline='') as f: reader = csv.DictReader(f) if column not in reader.fieldnames: raise KeyError(f"Column '{column}' not found. " f"Available columns: {reader.fieldnames}") for line_num, row in enumerate(reader, start=2): value = row[column].strip() if not value: # skip blank cells skipped += 1 continue try: total += float(value) except ValueError: print(f"Warning: skipping non-numeric value " f"'{value}' on line {line_num}") skipped += 1 except FileNotFoundError: print(f"Error: file '{filename}' not found.") return None if skipped: print(f"Skipped {skipped} row(s).") return total if __name__ == '__main__': result = sum_amount() if result is not None: print(f"Total amount: {result:.2f}") ``` ## Notes - **pandas** is cleanest but requires `pip install pandas`. - The **csv module** version has no external dependencies. - Make sure the column is literally named `amount` in the header (case-sensitive). Let me know your CSV's structure (delimiter, encoding, etc.) if you need adjustments.

Output

## Reading and Summing Sales Data Here's a clean Python solution with error handling: ```python import csv from pathlib import Path 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 Raises: FileNotFoundError: If the CSV file doesn't exist KeyError: If the 'amount' column is missing ValueError: If amount values can't be converted to numbers """ path = Path(filepath) if not path.exists(): raise FileNotFoundError(f"File not found: '{filepath}'") total = 0.0 row_count = 0 errors = [] with open(path, newline="", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile) # Check if '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 (row 1 = header) raw_value = row["amount"].strip() try: total += float(raw_value) row_count += 1 except ValueError: errors.append(f" Row {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).") return total if __name__ == "__main__": try: total = sum_sales_amount("sales.csv") print(f"💰 Total Sales Amount: ${total:,.2f}") except (FileNotFoundError, KeyError) as e: print(f"❌ Error: {e}") ``` --- ### Example `sales.csv` ``` id,product,amount 1,Widget A,150.00 2,Widget B,200.50 3,Widget C,99.99 4,Widget D,350.00 ``` ### Example Output ``` ✅ Processed 4 valid row(s). 💰 Total Sales Amount: $800.49 ``` --- ### Alternative: Using pandas (simpler) ```python import pandas as pd df = pd.read_csv("sales.csv") total = df["amount"].sum() print(f"Total Sales Amount: ${total:,.2f}") ``` --- ### Key Features | Feature | Description | |---|---| | `csv.DictReader` | Reads rows as dictionaries using header names | | `Path.exists()` | Safely checks if the file exists before opening | | Error handling | Skips invalid values instead of crashing | | `float()` conversion | Handles both integer and decimal amounts | | Formatted output | Uses `{:,.2f}` for currency formatting |

Output

```python import pandas as pd df = pd.read_csv('sales.csv') total = df['amount'].sum() print(total) ``` Done. Read CSV → sum amount column → print. Want file validation or error handling?

Output

Code normal (code rule): ```python import csv with open("sales.csv") as f: reader = csv.DictReader(f) total = sum(float(row["amount"]) for row in reader) print(f"Total: {total}") ``` Pandas way, shorter: ```python import pandas as pd total = pd.read_csv("sales.csv")["amount"].sum() print(f"Total: {total}") ``` Big file → use csv (stream, low mem). Else pandas fine.

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 csv total = 0 with open('sales.csv') as f: for row in csv.DictReader(f): total += float(row['amount']) print(total) ``` **Skipped:** error handling for missing file or malformed CSV, assume `amount` is always present and numeric. **Risk:** if `amount` has non-numeric values, it crashes; add `.strip()` and a try-except if the data is dirty.

Output

```python import csv with open("sales.csv") as f: total = sum(float(row["amount"]) for row in csv.DictReader(f)) print(total) ``` Skipped: error handling for missing file/bad rows, and didn't check your amount format (assumed plain numbers, no `$` or commas). If it has those, say so.

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 blank or non-numeric `amount` cells.

Output

# Rate Limiting in FastAPI Here are several approaches, from simple to advanced: ## 1. **Using `slowapi` (Recommended - Easiest)** ```bash pip install slowapi ``` ```python from fastapi import FastAPI from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from fastapi.responses import JSONResponse app = FastAPI() limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, lambda request, exc: 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"} ``` ## 2. **Using `aioredis` (For Distributed Systems)** ```bash pip install redis aioredis ``` ```python from fastapi import FastAPI, HTTPException, Request from redis import asyncio as aioredis import time app = FastAPI() redis_client = None @app.on_event("startup") async def startup(): global redis_client redis_client = await aioredis.from_url("redis://localhost") @app.on_event("shutdown") async def shutdown(): await redis_client.close() async def rate_limit_check(request: Request, max_requests: int, window: int): """ max_requests: number of allowed requests window: time window in seconds """ client_ip = request.client.host key = f"rate_limit:{client_ip}" current = await redis_client.incr(key) if current == 1: await redis_client.expire(key, window) if current > max_requests: raise HTTPException(status_code=429, detail="Rate limit exceeded") @app.get("/api/data") async def get_data(request: Request): await rate_limit_check(request, max_requests=5, window=60) return {"message": "Success"} ``` ## 3. **Simple In-Memory Rate Limiting (No Dependencies)** ```python from fastapi import FastAPI, HTTPException, Request from collections import defaultdict import time from typing import Dict, List app = FastAPI() # Store: {client_ip: [timestamp1, timestamp2, ...]} request_times: Dict[str, List[float]] = defaultdict(list) def rate_limit_check(client_ip: str, max_requests: int, window_seconds: int): """Simple in-memory rate limiter""" current_time = time.time() # Remove old requests outside the window request_times[client_ip] = [ req_time for req_time in request_times[client_ip] if current_time - req_time < window_seconds ] if len(request_times[client_ip]) >= max_requests: raise HTTPException( status_code=429, detail=f"Rate limit exceeded: {max_requests} requests per {window_seconds}s" ) request_times[client_ip].append(current_time) @app.get("/api/data") async def get_data(request: Request): client_ip = request.client.host rate_limit_check(client_ip, max_requests=5, window_seconds=60) return {"message": "Success"} @app.post("/api/submit") async def submit_data(request: Request, data: dict): client_ip = request.client.host rate_limit_check(client_ip, max_requests=10, window_seconds=3600) return {"status": "received"} ``` ## 4. **Advanced: Custom Decorator** ```python from functools import wraps from fastapi import FastAPI, HTTPException, Request from typing import Callable import time from collections import defaultdict app = FastAPI() request_tracker = defaultdict(list) def rate_limit(max_requests: int, window_seconds: int): """Decorator for rate limiting""" def decorator(func: Callable): @wraps(func) async def wrapper(request: Request, *args, **kwargs): client_ip = request.client.host current_time = time.time() # Clean old requests request_tracker[client_ip] = [ req_time for req_time in request_tracker[client_ip] if current_…

Output

# Rate Limiting in FastAPI The cleanest way to add rate limiting is with the **`slowapi`** library, which wraps the popular `limits` package. ## 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 # Limit based on client IP address limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @app.get("/items") @limiter.limit("5/minute") # max 5 requests per minute per IP async def read_items(request: Request): # 'request' param is REQUIRED return {"items": [1, 2, 3]} ``` > ⚠️ **Important:** The endpoint **must** include a `request: Request` parameter, or `slowapi` can't read the client address and will raise an error. ## Common Limit Syntax ```python @limiter.limit("100/day") @limiter.limit("10/hour") @limiter.limit("5/minute") @limiter.limit("2/second") @limiter.limit("5/minute;100/day") # multiple limits combined ``` ## Applying a Default Limit to All Endpoints ```python limiter = Limiter(key_func=get_remote_address, default_limits=["200/day", "50/hour"]) ``` ## Rate Limit Per User (instead of per IP) If you have authentication, limit by user ID rather than IP: ```python def get_user_id(request: Request) -> str: # however you identify your user (e.g. from a JWT / session) user = request.state.user return str(user.id) if user else get_remote_address(request) limiter = Limiter(key_func=get_user_id) ``` ## Production Note: Use Redis By default, limits are stored **in-memory**, which breaks if you run multiple workers/instances. Use a shared backend like Redis: ```bash pip install "slowapi[redis]" ``` ```python limiter = Limiter( key_func=get_remote_address, storage_uri="redis://localhost:6379", ) ``` --- ### Alternative: Manual Middleware If you'd rather not add a dependency, you can roll your own with Redis directly. Let me know if you want that version. To help me tailor this: **do you have user authentication, and are you running multiple workers/instances?** That determines whether you should limit by IP or user, and whether you need Redis.

Output

# Rate Limiting in FastAPI I'll show you a **production-ready rate limiting implementation** with multiple approaches, from simple to advanced. ## Setup ```bash pip install fastapi redis slowapi limits ``` ## Core Implementation ### `rate_limiter.py` — Reusable Rate Limiting Utilities ```python import time import asyncio from collections import defaultdict from dataclasses import dataclass, field from typing import Optional from functools import wraps import redis.asyncio as aioredis from fastapi import HTTPException, Request, status # ─── Data Classes ───────────────────────────────────────────────────────────── @dataclass class RateLimitConfig: requests: int # Max requests allowed window: int # Time window in seconds block_duration: int = 0 # Extra block time after limit hit (seconds) @dataclass class RateLimitState: count: int = 0 window_start: float = field(default_factory=time.time) blocked_until: float = 0 # ─── In-Memory Rate Limiter (single server / dev) ───────────────────────────── class InMemoryRateLimiter: """ Simple sliding-window rate limiter backed by local memory. ⚠️ Not suitable for multi-process/multi-server deployments. """ def __init__(self): self._store: dict[str, RateLimitState] = defaultdict(RateLimitState) self._lock = asyncio.Lock() async def is_allowed( self, key: str, config: RateLimitConfig, ) -> tuple[bool, dict]: """ Returns (allowed, headers_dict). headers_dict contains standard RateLimit-* response headers. """ async with self._lock: now = time.time() state = self._store[key] # Check if currently blocked if state.blocked_until > now: retry_after = int(state.blocked_until - now) return False, { "X-RateLimit-Limit": str(config.requests), "X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(int(state.blocked_until)), "Retry-After": str(retry_after), } # Reset window if expired if now - state.window_start >= config.window: state.count = 0 state.window_start = now # Check limit if state.count >= config.requests: if config.block_duration: state.blocked_until = now + config.block_duration reset_time = int(state.window_start + config.window) return False, { "X-RateLimit-Limit": str(config.requests), "X-RateLimit-Remaining": "0", "X-RateLimit-Reset": str(reset_time), "Retry-After": str(int(state.window_start + config.window - now)), } # Allow request state.count += 1 remaining = config.requests - state.count reset_time = int(state.window_start + config.window) return True, { "X-RateLimit-Limit": str(config.requests), "X-RateLimit-Remaining": str(remaining), "X-RateLimit-Reset": str(reset_time), } async def reset(self, key: str) -> None: async with self._lock: self._store.pop(key, None) # ─── Redis Rate Limiter (production / multi-server) ─────────────────────────── class RedisRateLimiter: """ Sliding-window rate limiter backed by Redis. Safe for multi-process and multi-server deployments. Uses an atomic Lua script to prevent race conditions. """ # Atomic Lua script: increment + set TTL in one round-trip _LUA_SCRIPT = """ local key = KEYS[1] local now = …

Output

# Rate Limiting for FastAPI ## Quick Setup Install: ```bash pip install slowapi ``` Add to endpoint: ```python from fastapi import FastAPI, HTTPException from slowapi import Limiter from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded app = FastAPI() limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, lambda r, e: HTTPException(429, "Too many requests")) @app.get("/api/endpoint") @limiter.limit("5/minute") async def my_endpoint(request: Request): return {"message": "ok"} ``` Done. 5 requests per minute per IP. Return 429 when hit. ## Customize Different rates per endpoint: - `"10/minute"` — 10 per minute - `"100/hour"` — 100 per hour - `"1000/day"` — 1000 per day Change key (user ID instead IP): ```python def get_user_id(request: Request): return request.headers.get("X-User-ID", "anonymous") @limiter.limit("20/minute", key_func=get_user_id) ``` Custom response: ```python def rate_limit_handler(request, exc): return JSONResponse( status_code=429, content={"error": "Rate limited. Try again later."} ) app.add_exception_handler(RateLimitExceeded, rate_limit_handler) ``` ## For Production slowapi stores in memory. Scale? Use Redis: ```bash pip install slowapi[redis] ``` ```python from slowapi.stores import RedisStore store = RedisStore("redis://localhost:6379") limiter = Limiter(key_func=get_remote_address, store=store) ``` Now limits sync across workers. Need different rules per user tier? Layer on auth check before limiter hits it.

Output

Use `slowapi`. Wrap FastAPI with limiter. ```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("/spammy") @limiter.limit("5/minute") async def spammy(request: Request): return {"ok": True} ``` Key points: - `request: Request` param required. slowapi read it. - Decorator order: `@app.get` top, `@limiter.limit` below. - Over limit → auto `429 Too Many Requests`. Prod note: `get_remote_address` use client IP. Behind proxy → set real IP via `X-Forwarded-For` or limiter break. Multi-worker → need shared `storage_uri` (Redis), else each worker own count.

Output

## Rate Limiting in FastAPI Best tool: **`slowapi`** (FastAPI-native, Redis-backed or in-memory). **Install:** ```bash pip install slowapi ``` **Add to app:** ```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") # ← set your limit here async def your_endpoint(request: Request): return {"message": "ok"} ``` 429 auto-return when limit hit. ## Limit Format | String | Meaning | |--------|---------| | `"10/minute"` | 10 req per min | | `"100/hour"` | 100 req per hr | | `"5/second"` | 5 req per sec | | `"10/minute;100/hour"` | Both, whichever hit first | ## Options **Per-user (auth token) instead of IP:** ```python def get_user_id(request: Request): return request.headers.get("X-API-Key", get_remote_address(request)) limiter = Limiter(key_func=get_user_id) ``` **Redis backend (persist across restarts/multiple workers):** ```python limiter = Limiter( key_func=get_remote_address, storage_uri="redis://localhost:6379" ) ``` ```bash pip install slowapi[redis] ``` ## Caveat IP-based rate limit bypass via proxy easy. For real prod: put nginx/Cloudflare rate limiting in front too.

Output

# Rate limit in 3 lines ```python from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) app = FastAPI() app.state.limiter = limiter @app.get("/endpoint") @limiter.limit("10/minute") async def my_endpoint(request: Request): return {"ok": True} ``` Install: `pip install slowapi` That's it. `slowapi` is the FastAPI standard. It returns 429 when over limit. If you need Redis for multi-process setups, add `redis` and pass `storage_uri="redis://localhost"` to `Limiter()`. --- **Skipped:** custom key functions (use IP by default), persistence config (in-memory works for one process), multiple limit tiers (add more decorators if you need them later). **Risk:** In-memory state resets on restart; for production load-balanced behind a reverse proxy, you need Redis or that proxy's rate limiting instead.

Output

Use `slowapi` — it's the standard FastAPI rate limiter and wraps this in a few lines. No need to hand-roll token buckets. ```python # pip install slowapi 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("/your-endpoint") @limiter.limit("5/minute") # tune the number async def your_endpoint(request: Request): # `request` param is required by slowapi ... ``` Three gotchas that will bite you: - **`request: Request` must be in the signature** or slowapi can't read the client. This is the #1 cause of "it silently doesn't limit." - **`get_remote_address` uses the socket IP.** Behind a proxy/load balancer that's the proxy, so everyone shares one limit. If you're behind nginx/cloudflare, swap the key_func to read `X-Forwarded-For` — and only trust that header if your proxy sets it, or clients will spoof it. - **Default limiter is in-memory**, so limits reset on restart and don't work across multiple workers/instances. For real multi-worker deploys pass `storage_uri="redis://..."`. --- Skipped: per-user (vs per-IP) limiting and custom 429 responses — ask if you need those. Didn't check whether you're behind a proxy or running multiple workers; both change the config above.

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 clearTimeout(timeoutId); // Set a new timeout timeoutId = setTimeout(() => { fn.apply(this, args); }, delay); }; } ``` ## Usage Examples ### Search Input ```javascript const handleSearch = (query) => { console.log('Searching for:', query); // API call here }; const debouncedSearch = debounce(handleSearch, 500); const input = document.querySelector('#search'); input.addEventListener('input', (e) => { debouncedSearch(e.target.value); }); ``` ### Window Resize ```javascript const handleResize = () => { console.log('Window resized'); }; const debouncedResize = debounce(handleResize, 250); window.addEventListener('resize', debouncedResize); ``` ### Form Input Validation ```javascript const validateEmail = (email) => { console.log('Validating:', email); }; const debouncedValidate = debounce(validateEmail, 300); document.querySelector('#email').addEventListener('change', (e) => { debouncedValidate(e.target.value); }); ``` ## Enhanced Version with Cancel & Flush For more control, here's an advanced version: ```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); }; // Execute immediately debounced.flush = () => { clearTimeout(timeoutId); fn.apply(this, arguments); }; return debounced; } ``` ### Using Cancel & Flush ```javascript const debouncedSearch = debounce(handleSearch, 500); // Cancel pending call debouncedSearch.cancel(); // Force immediate execution debouncedSearch.flush(); ``` ## Advanced Version with Leading & Trailing ```javascript function debounce(fn, delay, options = {}) { let timeoutId; let lastTime = 0; const { leading = false, trailing = true, maxWait } = options; const debounced = function(...args) { const now = Date.now(); // Handle leading edge if (leading && now - lastTime >= delay) { fn.apply(this, args); lastTime = now; return; } clearTimeout(timeoutId); timeoutId = setTimeout(() => { if (trailing) { fn.apply(this, args); } lastTime = Date.now(); }, delay); // Handle maxWait if (maxWait && now - lastTime >= maxWait) { clearTimeout(timeoutId); fn.apply(this, args); lastTime = now; } }; debounced.cancel = () => clearTimeout(timeoutId); debounced.flush = () => { clearTimeout(timeoutId); fn.apply(this, arguments); }; return debounced; } ``` ### Using with Options ```javascript // Execute on leading edge (first call) const debouncedSearch = debounce(handleSearch, 500, { leading: true }); // Guarantee execution at least every 2 seconds const debouncedResize = debounce(handleResize, 300, { trailing: true, maxWait: 2000 }); ``` ## Key Differences | Version | Features | Use Case | |---------|----------|----------| | **Basic** | Simple debounce | Search, validation | | **Cancel/Flush** | Control execution | Form submissions | | **Advanced** | Leading, trailing, maxWait | Complex interactions | Choose the version that fits your needs!

Output

# Debounce Function Here's a reusable `debounce` implementation in vanilla JavaScript: ```javascript function debounce(fn, delay) { let timeoutId; return function (...args) { // Preserve the calling context const context = this; // Cancel the previous scheduled call clearTimeout(timeoutId); // Schedule a new call after `delay` ms timeoutId = setTimeout(() => { fn.apply(context, args); }, delay); }; } ``` ## How It Works 1. **Closure over `timeoutId`** — Each debounced function gets its own private timer reference. 2. **`clearTimeout` on each call** — Cancels any pending execution, so rapid calls keep resetting the timer. 3. **`fn.apply(context, args)`** — Preserves `this` and forwards all arguments so the original function behaves normally. ## Basic Usage ```javascript function search(query) { console.log('Searching for:', query); } const debouncedSearch = debounce(search, 300); // Rapid calls — only the last one runs after 300ms debouncedSearch('a'); debouncedSearch('ap'); debouncedSearch('app'); // → logs "Searching for: app" once, 300ms after the last call ``` ## Real-World Example (Event Listener) ```javascript const input = document.querySelector('#search'); input.addEventListener( 'input', debounce((e) => { console.log('API call with:', e.target.value); }, 500) ); ``` --- ## Enhanced Version (with `immediate` and `cancel`) For more flexibility, here's a version supporting leading-edge execution and manual cancellation: ```javascript function debounce(fn, delay, immediate = false) { let timeoutId; function debounced(...args) { const context = this; const callNow = immediate && !timeoutId; clearTimeout(timeoutId); timeoutId = setTimeout(() => { timeoutId = null; if (!immediate) fn.apply(context, args); }, delay); if (callNow) fn.apply(context, args); } // Allow manual cancellation of pending calls debounced.cancel = function () { clearTimeout(timeoutId); timeoutId = null; }; return debounced; } ``` ### Usage of enhanced features ```javascript // Fire immediately on first call, then ignore for `delay` ms const saveNow = debounce(() => console.log('Saved!'), 1000, true); // Cancel a pending debounced call const handler = debounce(doSomething, 500); handler(); handler.cancel(); // doSomething will never run ``` **Key points:** - `immediate: true` runs `fn` on the **leading edge** instead of trailing. - `.cancel()` lets you abort a scheduled call (useful for cleanup in components).

Output

# Debounce Function in Vanilla JavaScript ## Core Implementation ```javascript /** * Creates a debounced version of a function that delays invoking it * until after `delay` milliseconds have elapsed since the last invocation. * * @param {Function} fn - The function to debounce * @param {number} delay - Milliseconds to wait after the last call * @returns {Function} - The debounced function */ function debounce(fn, delay) { let timerId = null; function debounced(...args) { // Clear any previously scheduled call clearTimeout(timerId); // Schedule a new call after the delay timerId = setTimeout(() => { fn.apply(this, args); // preserve `this` context and arguments timerId = null; }, delay); } // Cancel any pending invocation debounced.cancel = function () { clearTimeout(timerId); timerId = null; }; // Check if a call is currently pending debounced.isPending = function () { return timerId !== null; }; return debounced; } ``` --- ## Extended Version — with `immediate` option ```javascript /** * @param {Function} fn - The function to debounce * @param {number} delay - Milliseconds to wait * @param {boolean} immediate - If true, invoke fn on the LEADING edge instead * @returns {Function} */ function debounce(fn, delay, immediate = false) { let timerId = null; function debounced(...args) { const callNow = immediate && timerId === null; clearTimeout(timerId); timerId = setTimeout(() => { timerId = null; if (!immediate) { fn.apply(this, args); // trailing edge call } }, delay); if (callNow) { fn.apply(this, args); // leading edge call } } debounced.cancel = function () { clearTimeout(timerId); timerId = null; }; debounced.isPending = function () { return timerId !== null; }; return debounced; } ``` --- ## Usage Examples ### 1. Search Input (trailing edge — default) ```javascript const fetchResults = (query) => { console.log(`Fetching results for: "${query}"`); }; const debouncedFetch = debounce(fetchResults, 300); // Simulate rapid keystrokes — only the LAST call fires after 300ms debouncedFetch("j"); debouncedFetch("ja"); debouncedFetch("jav"); debouncedFetch("java"); // ← only this fires, 300ms after last keystroke ``` ### 2. Window Resize Handler ```javascript const handleResize = () => { console.log(`Window size: ${window.innerWidth}x${window.innerHeight}`); }; window.addEventListener("resize", debounce(handleResize, 200)); ``` ### 3. Immediate / Leading Edge (e.g. button click guard) ```javascript const saveData = () => console.log("Saved!"); const debouncedSave = debounce(saveData, 1000, true); // Fires immediately on first click, ignores clicks within the next 1000ms button.addEventListener("click", debouncedSave); ``` ### 4. Cancelling a Pending Call ```javascript const log = debounce((msg) => console.log(msg), 500); log("hello"); log.cancel(); // prevent the call before it fires console.log(log.isPending()); // false ``` ### 5. Preserving `this` Context (class methods) ```javascript class SearchBox { constructor() { this.query = ""; this.search = debounce(this.search, 300); } search(term) { this.query = term; // `this` is correctly bound console.log(`Searching: ${this.query}`); } } ``` --- ## How It Works — Visual Timeline ``` Rapid calls: |--A--B--C--D--| (all within delay window) Timer resets: ↺ ↺ ↺ ↺ Actual call: |--D (only last call executes) With immediate=true: First call: |A (fires immediately) Rapid calls: --B--C--D-- (all suppressed) After silence: ↺ (timer clears, ready for next burst) ``` --- ## Key Design Decisions | Concern | Solution | |---|---| | `this` context | `fn.apply(this, args)` preserves the caller's context | | Arguments | Rest params `...args` capture all arguments correctly | | Memory leaks | `cancel()` clears the timer and nulls the reference | | Observability | `…

Output

```javascript function debounce(fn, delay) { let timeoutId; return function(...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => fn(...args), delay); }; } ``` **How work:** - Store `timeoutId` in closure - Each call: clear old timeout, start new one - Only fire `fn` if no new call within `delay` ms **Usage:** ```javascript const handleSearch = debounce((query) => { console.log("Search:", query); }, 300); input.addEventListener("input", (e) => { handleSearch(e.target.value); }); ``` Type 3 chars fast → fire once at end, not 3 times. **Bonus — with context + return value:** ```javascript function debounce(fn, delay) { let timeoutId; return function(...args) { clearTimeout(timeoutId); return new Promise((resolve) => { timeoutId = setTimeout(() => resolve(fn.apply(this, args)), delay); }); }; } ``` Keep `this` binding + Promise for awaitable debounce.

Output

Debounce = delay fn til calls stop. Clear old timer each call. ```javascript function debounce(fn, delay) { let timer; return function (...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); }; } ``` `this` bind + `args` spread = work on methods too. Each call reset timer → only last fire after `delay`. Usage: ```javascript const onResize = debounce(() => console.log('resized'), 200); window.addEventListener('resize', onResize); ```

Output

```javascript function debounce(fn, delay) { let timer; return function (...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); }; } ``` **How work:** - `timer` persist in closure across calls - Each call kill old timer, start new - Only last call survive full `delay` → fires `fn` - `apply(this, args)` preserve context + spread args **Use:** ```javascript const onResize = debounce(() => console.log('resized'), 300); window.addEventListener('resize', onResize); // Cancel if need: 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(); // stop pending call ``` **Edge cases handled:** | Case | Behavior | |------|----------| | Rapid calls | Only last fires | | `this` context | Preserved via `apply` | | Arg passing | All args forwarded | | Cancel | `.cancel()` clears pending |

Output

```javascript function debounce(fn, delay) { let timeoutId; return function(...args) { clearTimeout(timeoutId); timeoutId = setTimeout(() => fn(...args), delay); }; } ``` **Usage:** ```javascript const search = debounce((query) => console.log('Searching:', query), 300); search('a'); search('ab'); search('abc'); // only this fires, 300ms later ``` **Skipped:** cancel method, leading/trailing edge options, `this` binding (use arrow functions or `.bind()` at call site). **Risk:** if `fn` throws, the error fires in the timeout; wrap it if you need different error handling.

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 d = debounce(() => calls++, 50); d(); d(); d(); setTimeout(() => console.assert(calls === 1, `expected 1, got ${calls}`), 80); ``` Skipped: cancel/flush methods and leading-edge option (not asked). `this` and args are preserved so it works on methods and events.

Output

```js function debounce(fn, delay) { let timer; return function (...args) { clearTimeout(timer); timer = setTimeout(() => fn.apply(this, args), delay); }; } ``` **Self-check:** ```js let count = 0; const inc = debounce(() => count++, 50); inc(); inc(); inc(); setTimeout(() => console.assert(count === 1, "should fire once"), 100); ``` --- **Skipped:** leading-edge firing, cancel/flush methods, max-wait — not asked for. **Risk:** `timer` is captured per returned function, so each call to `debounce()` is independent — don't share the returned function across contexts expecting separate timers.

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 default 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 handleStart = () => setIsActive(true); const handlePause = () => setIsActive(false); const handleReset = () => { setSeconds(initialSeconds); setIsActive(false); }; return ( <div style={styles.container}> <h2>Countdown Timer</h2> <div style={styles.display}>{seconds}s</div> <div style={styles.buttonGroup}> <button onClick={handleStart} disabled={isActive}> Start </button> <button onClick={handlePause} disabled={!isActive}> Pause </button> <button onClick={handleReset}>Reset</button> </div> </div> ); } const styles = { container: { textAlign: 'center', padding: '20px', fontFamily: 'Arial, sans-serif', }, display: { fontSize: '48px', fontWeight: 'bold', margin: '20px 0', color: '#333', }, buttonGroup: { display: 'flex', gap: '10px', justifyContent: 'center', }, }; ``` ## Advanced Timer with Formatted Display ```jsx import { useState, useEffect, useCallback } from 'react'; export default function AdvancedCountdownTimer({ initialSeconds = 3600, onComplete = null, showMilliseconds = false }) { const [timeLeft, setTimeLeft] = useState(initialSeconds); const [isActive, setIsActive] = useState(false); useEffect(() => { let interval = null; if (isActive && timeLeft > 0) { interval = setInterval(() => { setTimeLeft(prev => { if (prev <= 1) { setIsActive(false); onComplete?.(); return 0; } return prev - 1; }); }, 1000); } return () => clearInterval(interval); }, [isActive, timeLeft, onComplete]); const formatTime = useCallback((seconds) => { const hours = Math.floor(seconds / 3600); const minutes = Math.floor((seconds % 3600) / 60); const secs = seconds % 60; const parts = []; if (hours > 0) parts.push(String(hours).padStart(2, '0')); parts.push(String(minutes).padStart(2, '0')); parts.push(String(secs).padStart(2, '0')); return parts.join(':'); }, []); const toggleTimer = () => setIsActive(!isActive); const reset = () => { setTimeLeft(initialSeconds); setIsActive(false); }; const isFinished = timeLeft === 0; return ( <div style={styles.advancedContainer}> <h2>Advanced Timer</h2> <div style={{ ...styles.advancedDisplay, color: isFinished ? '#e74c3c' : '#2ecc71', }}> {formatTime(timeLeft)} </div> <div style={styles.progressBar}> <div style={{ ...styles.progressFill, width: `${(timeLeft / initialSeconds) * 100}%`, }} /> </div> <div style={styles.buttonGroup}> <button onClick={toggleTimer} style={{ ...styles.button, backgroundColor: isActive ? '#e74c3c' : '#2ecc71', }} > {isActive ? 'Pause' : 'Start'} </button> <button onClick={reset} style={styles.button}> Reset </button> </div> {isFinished && <div style={styles.finishedMessage}>⏰ Time's up!</div>} </div> ); } const styles = { advancedContainer: { textAlign: 'center', padding: '30px', maxWidth: '400px', margin: '0 auto', backgroundColor: '#f8f9fa', borderRadius: '10px', boxShadow: '0 2px 10px rgba…

Output

# Countdown Timer Component Here's a flexible countdown timer component in React with start, pause, and reset functionality. ## Basic Component ```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) => { if (prev <= 1) { clearTimer(); setIsRunning(false); onComplete?.(); return 0; } return prev - 1; }); }, 1000); } // Cleanup on unmount or when dependencies change return clearTimer; }, [isRunning, secondsLeft, clearTimer, onComplete]); const handleStart = () => { if (secondsLeft > 0) setIsRunning(true); }; const handlePause = () => setIsRunning(false); const handleReset = () => { setIsRunning(false); setSecondsLeft(initialSeconds); }; // Format seconds into 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={90} onComplete={() => alert('Time is up!')} /> ); } ``` ## Optional Styling ```css .countdown-timer { display: flex; flex-direction: column; align-items: center; gap: 1rem; font-family: sans-serif; } .time-display { font-size: 3rem; font-weight: bold; font-variant-numeric: tabular-nums; } .controls button { padding: 0.5rem 1.25rem; margin: 0 0.25rem; font-size: 1rem; cursor: pointer; border: none; border-radius: 6px; background: #4f46e5; color: white; } .controls button:disabled { background: #9ca3af; cursor: not-allowed; } ``` ## Key Features | Feature | Description | |---------|-------------| | **Start/Pause/Reset** | Full playback control | | **`onComplete` callback** | Runs when timer hits zero | | **MM:SS formatting** | Clean display for longer durations | | **Cleanup** | Prevents memory leaks via `useEffect` cleanup | | **`tabular-nums`** | Prevents digits from shifting as they change | ## Notes & Caveats - **Timing drift:** `setInterval` isn't perfectly accurate and drifts when the tab is backgrounded. For precise timing, compare against `Date.now()` instead of decrementing. Here's the drift-resistant version if you need accuracy: ```jsx useEffect(() => { if (!isRunning) return; const endTime = Date.now() + secondsLeft * 1000; const tick = () => { const remaining = Math.round((endTime - Date.now()) / 1000); if (remaining <= 0) { setSecondsLeft(0); setIsRunning(false); onComplete?.(); } else { setSecondsLeft(remaining); } }; intervalRef.current = setInterval(tick, 250); // check more frequently return clearTimer; }, [isRunning]); // note: fewer dependencies ``` Want me to add anything like a circular progress ring, custom input for setting the time, or sound on completion?

Output

# Countdown Timer Component ## Component Code ```tsx import React, { useState, useEffect, useRef, useCallback } from "react"; interface CountdownTimerProps { initialSeconds?: number; } const CountdownTimer: React.FC<CountdownTimerProps> = ({ initialSeconds = 60, }) => { const [timeLeft, setTimeLeft] = useState(initialSeconds); const [isRunning, setIsRunning] = useState(false); const [inputSeconds, setInputSeconds] = useState(initialSeconds.toString()); const [isFinished, setIsFinished] = useState(false); const intervalRef = useRef<ReturnType<typeof setInterval> | null>(null); const clearTimer = useCallback(() => { if (intervalRef.current) { clearInterval(intervalRef.current); intervalRef.current = null; } }, []); useEffect(() => { if (isRunning) { intervalRef.current = setInterval(() => { setTimeLeft((prev) => { if (prev <= 1) { clearTimer(); setIsRunning(false); setIsFinished(true); return 0; } return prev - 1; }); }, 1000); } return () => clearTimer(); }, [isRunning, clearTimer]); const handleStart = () => { if (timeLeft > 0) { setIsFinished(false); setIsRunning(true); } }; const handlePause = () => { setIsRunning(false); clearTimer(); }; const handleReset = () => { clearTimer(); setIsRunning(false); setIsFinished(false); const parsed = parseInt(inputSeconds, 10); const seconds = isNaN(parsed) || parsed <= 0 ? initialSeconds : parsed; setTimeLeft(seconds); }; const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>) => { setInputSeconds(e.target.value); }; const handleInputSubmit = (e: React.FormEvent) => { e.preventDefault(); const parsed = parseInt(inputSeconds, 10); if (!isNaN(parsed) && parsed > 0) { clearTimer(); setIsRunning(false); setIsFinished(false); setTimeLeft(parsed); } }; const formatTime = (seconds: number): string => { const hrs = Math.floor(seconds / 3600); const mins = Math.floor((seconds % 3600) / 60); const secs = seconds % 60; if (hrs > 0) { return `${String(hrs).padStart(2, "0")}:${String(mins).padStart(2, "0")}:${String(secs).padStart(2, "0")}`; } return `${String(mins).padStart(2, "0")}:${String(secs).padStart(2, "0")}`; }; const totalSeconds = parseInt(inputSeconds, 10) || initialSeconds; const progressPercentage = totalSeconds > 0 ? ((totalSeconds - timeLeft) / totalSeconds) * 100 : 0; const getTimerColor = () => { if (isFinished) return "#ef4444"; if (timeLeft <= 10) return "#f97316"; if (timeLeft <= 30) return "#eab308"; return "#3b82f6"; }; const timerColor = getTimerColor(); return ( <div style={styles.wrapper}> <div style={styles.card}> <h1 style={styles.title}>Countdown Timer</h1> {/* Circular Progress */} <div style={styles.circleContainer}> <svg width="200" height="200" style={styles.svg}> {/* Background circle */} <circle cx="100" cy="100" r="85" fill="none" stroke="#e5e7eb" strokeWidth="10" /> {/* Progress circle */} <circle cx="100" cy="100" r="85" fill="none" stroke={timerColor} strokeWidth="10" strokeLinecap="round" strokeDasharray={`${2 * Math.PI * 85}`} strokeDashoffset={`${2 * Math.PI * 85 * (progressPercentage / 100)}`} transform="rotate(-90 100 100)" style={styles.progressCircle} /> </svg> <div style={styles.timeDisplay}> <span style={{ ...styles.timeText, color: timerColor }}> {formatTime(timeLeft)} </span> {isFinished && <span style={styles.finishedText}>Time's up!</span>} </div> …

Output

```jsx import { useState, useEffect } from 'react'; export default 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]); useEffect(() => { if (seconds === 0) { setIsActive(false); } }, [seconds]); const toggle = () => setIsActive(!isActive); const reset = () => { setIsActive(false); setSeconds(initialSeconds); }; const minutes = Math.floor(seconds / 60); const displaySeconds = seconds % 60; return ( <div style={{ textAlign: 'center', padding: '20px' }}> <div style={{ fontSize: '48px', fontWeight: 'bold', margin: '20px 0' }}> {String(minutes).padStart(2, '0')}:{String(displaySeconds).padStart(2, '0')} </div> <button onClick={toggle} style={{ margin: '5px', padding: '10px 20px' }}> {isActive ? 'Pause' : 'Start'} </button> <button onClick={reset} style={{ margin: '5px', padding: '10px 20px' }}> Reset </button> </div> ); } ``` ## Use: ```jsx <CountdownTimer initialSeconds={120} /> ``` ## Key: - `useEffect` handle interval. Stop at 0 or pause. - MM:SS format. `padStart` add leading zeros. - Two buttons: toggle start/pause, reset. - Auto-cleanup interval on unmount/dependency change.

Output

```jsx import { useState, useEffect, useRef } from 'react'; function CountdownTimer({ seconds = 60, onComplete }) { const [timeLeft, setTimeLeft] = useState(seconds); const [running, setRunning] = useState(false); const intervalRef = useRef(null); useEffect(() => { if (!running) return; intervalRef.current = setInterval(() => { setTimeLeft((t) => { if (t <= 1) { clearInterval(intervalRef.current); setRunning(false); onComplete?.(); return 0; } return t - 1; }); }, 1000); return () => clearInterval(intervalRef.current); }, [running, onComplete]); const start = () => timeLeft > 0 && setRunning(true); const pause = () => setRunning(false); const reset = () => { setRunning(false); setTimeLeft(seconds); }; const mins = String(Math.floor(timeLeft / 60)).padStart(2, '0'); const secs = String(timeLeft % 60).padStart(2, '0'); return ( <div> <span>{mins}:{secs}</span> <button onClick={start} disabled={running}>Start</button> <button onClick={pause} disabled={!running}>Pause</button> <button onClick={reset}>Reset</button> </div> ); } export default CountdownTimer; ``` Key points caveman: - `useRef` hold interval id → clean stop, no stale closure. - Functional `setTimeLeft` → always fresh value. - Cleanup in effect return → no leak on unmount. - `onComplete?.()` fire at zero. Use: `<CountdownTimer seconds={120} onComplete={() => alert('done')} />`

Output

```jsx import { useState, useEffect, useRef } from "react"; export default function CountdownTimer({ initialSeconds = 60 }) { const [seconds, setSeconds] = useState(initialSeconds); const [running, setRunning] = useState(false); const intervalRef = useRef(null); useEffect(() => { if (running && seconds > 0) { intervalRef.current = setInterval(() => { setSeconds((s) => s - 1); }, 1000); } else if (seconds === 0) { setRunning(false); } return () => clearInterval(intervalRef.current); }, [running, seconds]); const reset = () => { setRunning(false); setSeconds(initialSeconds); }; const fmt = (s) => { const m = Math.floor(s / 60).toString().padStart(2, "0"); const sec = (s % 60).toString().padStart(2, "0"); return `${m}:${sec}`; }; return ( <div style={{ textAlign: "center", fontFamily: "monospace" }}> <h1 style={{ fontSize: "4rem", margin: "1rem 0" }}> {fmt(seconds)} </h1> <div style={{ display: "flex", gap: "0.5rem", justifyContent: "center" }}> <button onClick={() => setRunning((r) => !r)} disabled={seconds === 0}> {running ? "Pause" : "Start"} </button> <button onClick={reset}>Reset</button> </div> {seconds === 0 && <p>⏰ Time's up!</p>} </div> ); } ``` **Usage:** ```jsx <CountdownTimer initialSeconds={90} /> ``` **Key points:** - `useEffect` deps on `running` + `seconds` — interval cleared/restarted on each state change, no stale closure bugs - `return () => clearInterval(...)` — cleanup prevents multiple concurrent intervals - `fmt()` — pads to `MM:SS` display - Start button disables at zero; reset restores `initialSeconds` **To extend:** - Custom end callback → add `onComplete` prop, call in `seconds === 0` branch - User-editable input → controlled input that sets `initialSeconds` via lifted state - Audio alert → `new Audio(...).play()` in the zero branch

Output

# Countdown Timer ```jsx export function CountdownTimer({ seconds }) { const [remaining, setRemaining] = React.useState(seconds); React.useEffect(() => { if (remaining <= 0) return; const id = setInterval(() => setRemaining(r => r - 1), 1000); return () => clearInterval(id); }, [remaining]); return <div>{remaining}s</div>; } ``` **Usage:** ```jsx <CountdownTimer seconds={30} /> ``` --- **Skipped:** pause/resume, completion callback, formatting (mm:ss), styling, reset button, sound effects. **Risk:** if `seconds` prop changes mid-countdown, timer restarts from the new value. Add `React.useEffect` dependency on a stable ID if that's wrong for your use case.

Output

```jsx import { useState, useEffect } from "react"; export default 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>; } ``` Skipped (nobody asked): start/pause/reset controls, mm:ss formatting, onComplete callback. Say the word and I'll add. Risk: `setInterval` drifts over long runs; fine for a display timer, not for timing-critical use.

Output

Here's the minimal version: ```jsx // CountdownTimer.jsx import { useState, useEffect } from "react"; export default function CountdownTimer({ seconds }) { const [remaining, setRemaining] = useState(seconds); useEffect(() => { if (remaining <= 0) return; const id = setInterval(() => setRemaining((r) => r - 1), 1000); return () => clearInterval(id); }, [remaining]); return <div>{remaining}s</div>; } ``` **Usage:** ```jsx <CountdownTimer seconds={60} /> ``` The interval re-registers every tick, which is negligible overhead but keeps the logic simple. For a long-running or high-frequency timer, switch to a single `setInterval` with a `ref` for the remaining value. --- **Skipped:** pause/reset controls, formatting as MM:SS, onComplete callback — add any if you need them. **Risk:** changing the `seconds` prop after mount won't reset the timer; add `key={seconds}` on the parent if you need that.

Output

# Email Validation Function Here's a practical email validation function using regex: ```python import re def validate_email(email: str) -> bool: """ Validates an email address using regex pattern. Args: email (str): The email address to validate Returns: bool: True if email is valid, False otherwise """ # Basic email regex pattern pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$' return re.match(pattern, email) is not None # Test cases test_emails = [ "user@example.com", # ✓ Valid "john.doe@company.co.uk", # ✓ Valid "test+tag@domain.org", # ✓ Valid "invalid.email@", # ✗ Invalid "missing@domain", # ✗ Invalid "spaces in@email.com", # ✗ Invalid "@example.com", # ✗ Invalid "user@.com", # ✗ Invalid ] for email in test_emails: result = validate_email(email) status = "✓" if result else "✗" print(f"{status} {email}") ``` --- ## More Robust Version For production use, consider using the `email-validator` library: ```python from email_validator import validate_email, EmailNotValidError def validate_email_robust(email: str) -> bool: """ Validates email address and checks if format is correct. Requires: pip install email-validator Args: email (str): The email address to validate Returns: bool: True if email is valid, False otherwise """ try: # Validate and normalize the email address valid = validate_email(email) return True except EmailNotValidError: return False # Test print(validate_email_robust("user@example.com")) # True print(validate_email_robust("invalid@")) # False ``` --- ## Comparison | Method | Pros | Cons | |--------|------|------| | **Regex** | No dependencies, fast | Limited accuracy | | **email-validator** | Comprehensive, RFC compliant | Requires installation | **Choose regex** for simple validation, **choose email-validator** for production systems.

Output

# Email Validation in Python Here are a few approaches depending on your needs: ## Option 1: Simple Regex (Good for most cases) ```python import re def validate_email(email): """ Validate an email address using a regular expression. Args: email (str): The email address to validate. Returns: bool: 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: Using the `email-validator` Library (Recommended for production) This library handles edge cases, internationalized domains, and can optionally check deliverability. ```bash pip install email-validator ``` ```python from email_validator import validate_email, EmailNotValidError def is_valid_email(email, check_deliverability=False): """ Validate an email address using the email-validator library. Args: email (str): The email address to validate. check_deliverability (bool): If True, checks if the domain can receive mail. Returns: str | None: The normalized email if valid, None otherwise. """ try: result = validate_email(email, check_deliverability=check_deliverability) return result.normalized except EmailNotValidError: return None ``` ## Usage Examples ```python # Option 1 print(validate_email("user@example.com")) # True print(validate_email("invalid.email")) # False print(validate_email("name@sub.domain.org")) # True print(validate_email("@example.com")) # False # Option 2 print(is_valid_email("user@example.com")) # user@example.com print(is_valid_email("not-an-email")) # None ``` ## Which should you use? | Use Case | Recommendation | |----------|---------------| | Quick validation, learning | Option 1 (regex) | | Production applications | Option 2 (library) | | Need to verify domain exists | Option 2 with `check_deliverability=True` | **Important note:** No regex can *perfectly* validate every email address according to the full RFC 5322 specification. The only way to truly confirm an email works is to **send a confirmation email**. The methods above check *format* validity, which is sufficient for most use cases. Would you like me to tailor this for a specific framework (Django, Flask, FastAPI) or add additional checks?

Output

```python import re def validate_email(email: str) -> bool: """ Validate email address using regex. 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 ``` **Usage:** ```python validate_email("user@example.com") # True validate_email("invalid.email@") # False validate_email("test+tag@domain.co.uk") # True ``` **What it check:** - Local part: letters, numbers, `._%+-` - `@` separator - Domain: letters, numbers, `.` `-` - TLD: min 2 letters **Better version** (RFC 5322 compliant): ```python from email_validator import validate_email as validate_email_lib def validate_email(email: str) -> bool: try: validate_email_lib(email) return True except Exception: return False ``` Install: `pip install email-validator` Regex OK for basic case. Library OK for strict validation + DNS check.

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 cover common case. For strict RFC 5322 use `email-validator` lib.

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 addresses, real domain existence, quoted local parts (RFC 5321 edge cases). Need stricter? Use `email-validator` lib: ```bash pip install email-validator ``` ```python from email_validator import validate_email, EmailNotValidError def validate_email_strict(email: str) -> bool: try: validate_email(email) return True except EmailNotValidError: return False ```

Output

```python import re def validate_email(email): return re.match(r'^[^@]+@[^@]+\.[^@]+$', email) is not None ``` **What I skipped:** RFC 5322 compliance (overkill for most apps), DNS checks, disposable domain lists, internationalized addresses, plus-addressing rules. Use this if you're checking user signups or forms. If you need bulletproof validation (payment, legal), switch to the `email-validator` package. **Risk:** The regex accepts `a@b.c` but rejects nothing with the format `x@y.z`. If you're sending emails, the only real test is "can you receive a reply?"—validate by sending a confirmation link.

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.co") assert not is_valid_email("a@b") assert not is_valid_email("a b@c.com") assert not is_valid_email("@c.com") ``` Skipped full RFC 5322 compliance (hundreds of lines, rejects some valid addresses anyway). This catches typos, not whether the address actually receives mail. Risk: if you need real validation, send a confirmation email instead — that's the only check that counts.

Output

```python import re _EMAIL_RE = re.compile(r"[^@]+@[^@]+\.[^@]+") def is_valid_email(email: str) -> bool: return bool(_EMAIL_RE.match(email)) ``` This catches the 99% case. For real validation, send a confirmation email — regex can't check deliverability. **Skipped:** RFC 5322 edge cases (quoted locals, IP literals, comments). **Risk:** accepts `a@b.c` style strings that aren't real domains; rejects nothing dangerous.