Async Validation — Complete Guide
In this tutorial, you'll learn about Async Validation. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Async validation performs checks that require I/O operations: database queries for uniqueness, external API calls for verification, and file system checks.
What You'll Learn
By the end of this lesson, you will implement async validators, batch database lookups to avoid N+1, and handle timeout and error cases.
Why It Matters
Many validation rules require data from external sources. Checking username uniqueness, email existence, or referral code validity requires async I/O that must complete before the request proceeds.
Real-World Use
User registration validates email uniqueness against the database, checks if the IP is in a blocklist (external API), and verifies the referral code exists — all async.
Async Validation Flow
sequenceDiagram
Pipeline->>Validator: Check email
Validator->>DB: SELECT COUNT(*) WHERE email=?
DB->>Validator: 1 (exists)
Validator->>Pipeline: Error: already taken
Pipeline->>Validator: Check username
Validator->>DB: SELECT COUNT(*) WHERE username=?
DB->>Validator: 0 (available)
Validator->>Pipeline: OK
Async Uniqueness Check
# async_unique.py
import asyncio
from typing import Any, Dict, List, Optional
class AsyncDatabase:
def __init__(self):
self.users = {
"alice@example.com": {"id": 1, "username": "alice"},
"bob@example.com": {"id": 2, "username": "bob"},
}
async def exists_by_email(self, email: str) -> bool:
await asyncio.sleep(0.01)
return email in self.users
async def exists_by_username(self, username: str) -> bool:
await asyncio.sleep(0.01)
return any(u["username"] == username for u in self.users.values())
class AsyncValidator:
def __init__(self, db: AsyncDatabase):
self.db = db
self.errors: List[str] = []
async def check_email_unique(self, email: str):
if await self.db.exists_by_email(email):
self.errors.append(f"Email '{email}' is already registered")
async def check_username_unique(self, username: str):
if await self.db.exists_by_username(username):
self.errors.append(f"Username '{username}' is taken")
async def validate_registration(self, email: str, username: str) -> List[str]:
self.errors = []
await asyncio.gather(
self.check_email_unique(email),
self.check_username_unique(username),
)
return self.errors
async def main():
db = AsyncDatabase()
validator = AsyncValidator(db)
errors = await validator.validate_registration("new@example.com", "newuser")
print(f"New user: errors={errors}")
errors2 = await validator.validate_registration("alice@example.com", "bob")
print(f"Existing user: errors={errors2}")
asyncio.run(main())
Expected output:
New user: errors=[]
Existing user: errors=["Email 'alice@example.com' is already registered", "Username 'bob' is taken"]
Batched Async Validation
# batched_async.py
import asyncio
from typing import Any, Dict, List, Set
class BatchedValidator:
def __init__(self):
self.pending_emails: Set[str] = set()
self.results: Dict[str, bool] = {}
async def check_email(self, email: str) -> bool:
self.pending_emails.add(email)
if len(self.pending_emails) >= 10:
await self._flush()
if email not in self.results:
await self._flush()
return not self.results.get(email, False)
async def _flush(self):
if not self.pending_emails:
return
batch = list(self.pending_emails)
self.pending_emails.clear()
await asyncio.sleep(0.02)
for email in batch:
self.results[email] = email.endswith("@example.com")
async def flush(self):
await self._flush()
async def main():
validator = BatchedValidator()
emails = ["a@x.com", "b@test.com", "c@example.com"]
for email in emails:
available = await validator.check_email(email)
print(f" Email {email:20s} available={available}")
await validator.flush()
asyncio.run(main())
Expected output:
Email a@x.com available=True
Email b@test.com available=True
Email c@example.com available=False
Timeout Handling
# timeout_validation.py
import asyncio
from typing import Any, Dict, List, Optional
class TimeoutValidator:
def __init__(self, timeout_seconds: float = 2.0):
self.timeout = timeout_seconds
async def _slow_check(self, value: str) -> bool:
await asyncio.sleep(5)
return True
async def check_with_timeout(self, value: str) -> Dict:
try:
result = await asyncio.wait_for(
self._slow_check(value),
timeout=self.timeout,
)
return {"valid": result, "source": "check"}
except asyncio.TimeoutError:
return {"valid": True, "source": "timeout_default"}
async def main():
validator = TimeoutValidator(timeout_seconds=0.5)
result = await validator.check_with_timeout("test")
print(f"Result: {result}")
asyncio.run(main())
Expected output:
Result: {'valid': True, 'source': 'timeout_default'}
Common Mistakes
1. Sequential Async Validators
Running async validators sequentially instead of parallel with asyncio.gather slows the pipeline.
2. No Timeout
An external validation API that hangs blocks the entire request. Always set timeouts.
3. Ignoring Results
Starting async validators without awaiting them means validation never completes.
4. N+1 Database Queries
Validating 10 items triggers 10 separate queries. Batch lookups into single queries.
5. Caching Validation Results
Uniqueness checks can be cached briefly. Subsequent checks in the same request reuse the result.
Practice Questions
1. When is async validation needed?
For database uniqueness checks, external API verifications, file system checks, or any I/O operation.
2. How do you run async validators in parallel?
Use asyncio.gather() to run all async checks concurrently.
3. Why set timeouts on async validators?
Slow external services should not block the entire request. Default to valid after timeout.
4. How do you batch async lookups?
Collect pending keys and flush them in a single query when the batch reaches a threshold.
Challenge
Build an async registration validator that checks email uniqueness (DB), username availability (DB), referral code validity (DB + batch), and IP reputation (external API with timeout).
FAQ
Mini Project: Async Validator Runner
# async_runner.py
import asyncio
from typing import Any, Callable, Dict, List
class AsyncValidatorRunner:
def __init__(self):
self.validators: List[Callable] = []
def add(self, fn: Callable):
self.validators.append(fn)
async def run(self, value: Any) -> List[str]:
results = await asyncio.gather(*[fn(value) for fn in self.validators],
return_exceptions=True)
errors = []
for r in results:
if isinstance(r, Exception):
errors.append(str(r))
elif r:
errors.append(r)
return errors
async def main():
runner = AsyncValidatorRunner()
async def check_db(v):
await asyncio.sleep(0.01)
return None if v != "taken" else "already exists"
async def check_api(v):
await asyncio.sleep(0.01)
return None
runner.add(check_db)
runner.add(check_api)
print(await runner.run("new"))
print(await runner.run("taken"))
asyncio.run(main())
Expected output:
[]
['already exists']
What's Next
You understand async validation. Next, learn the Express validation pipeline, then explore Joi validation.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro