Documentation Index
Fetch the complete documentation index at: https://mintlify.com/Zozi96/hash-forge/llms.txt
Use this file to discover all available pages before exploring further.
Hash Forge provides a full async API for every core operation. Under the hood, each async method calls asyncio.get_running_loop().run_in_executor(None, ...) to offload the CPU-bound hash computation to a thread-pool executor. This means the event loop is never blocked while a hash is being computed — your server stays responsive to other requests even during expensive Argon2 or bcrypt operations.
All async methods are inherited from AsyncHashMixin. Every HashManager instance includes them automatically — no additional setup or subclassing is required.
Available Async Methods
| Method | Signature | Description |
|---|
hash_async | (string: str) -> str | Hash with the preferred hasher, non-blocking |
verify_async | (string: str, hashed_string: str) -> bool | Verify against a stored hash, non-blocking |
needs_rehash_async | (hashed_string: str) -> bool | Check if rehash is needed, non-blocking |
hash_many_async | (strings: list[str]) -> list[str] | Hash a list of strings concurrently |
verify_many_async | (pairs: list[tuple[str, str]]) -> list[bool] | Verify a list of pairs concurrently |
All async methods run the underlying synchronous operation in a thread-pool executor via asyncio.get_running_loop().run_in_executor. This keeps the asyncio event loop free for I/O tasks while the CPU-bound hash work runs on a worker thread.
FastAPI Integration
The example below shows register and login endpoints that use hash_async and verify_async so neither endpoint blocks the event loop.
import asyncio
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from hash_forge import HashManager
app = FastAPI(title="Hash Forge Auth API", version="1.0.0")
# Initialize HashManager with Argon2 (recommended for web apps)
hash_manager = HashManager.from_algorithms("argon2")
# In-memory user database (use a real DB in production)
users_db: dict[str, dict] = {}
class UserRegister(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
password: str = Field(..., min_length=8)
email: str
class UserLogin(BaseModel):
username: str
password: str
class UserResponse(BaseModel):
username: str
email: str
message: str
@app.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register(user: UserRegister):
"""Register a new user with async password hashing."""
if user.username in users_db:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already exists",
)
# Hash password asynchronously — does not block the event loop
password_hash = await hash_manager.hash_async(user.password)
users_db[user.username] = {
"email": user.email,
"password_hash": password_hash,
}
return UserResponse(
username=user.username,
email=user.email,
message="User registered successfully",
)
@app.post("/login", response_model=UserResponse)
async def login(credentials: UserLogin):
"""Login user with async password verification."""
user = users_db.get(credentials.username)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid username or password",
)
# Verify password asynchronously — does not block the event loop
is_valid = await hash_manager.verify_async(
credentials.password,
user["password_hash"],
)
if not is_valid:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid username or password",
)
# Transparently upgrade the hash if parameters changed
needs_rehash = await hash_manager.needs_rehash_async(user["password_hash"])
if needs_rehash:
new_hash = await hash_manager.hash_async(credentials.password)
users_db[credentials.username]["password_hash"] = new_hash
return UserResponse(
username=credentials.username,
email=user["email"],
message="Login successful",
)
Batch Hashing
hash_many_async hashes a list of strings concurrently. Internally it builds a list of hash_async coroutines and runs them together with asyncio.gather, so all hashes are computed in parallel on the thread pool.
import asyncio
from hash_forge import HashManager
async def batch_example():
manager = HashManager.from_algorithms("pbkdf2_sha256")
passwords = ["user1_pass", "user2_pass", "user3_pass", "user4_pass"]
# All four hashes computed concurrently
hashes = await manager.hash_many_async(passwords)
for password, hash_value in zip(passwords, hashes):
print(f"{password} -> {hash_value[:50]}...")
# Verify multiple pairs concurrently
pairs = [
("user1_pass", hashes[0]),
("user2_pass", hashes[1]),
("wrong_password", hashes[2]), # will be False
]
results = await manager.verify_many_async(pairs)
print(results) # [True, True, False]
asyncio.run(batch_example())
batch-register endpoint below shows this pattern applied inside a FastAPI route:
@app.post("/batch-register")
async def batch_register(users: list[UserRegister]):
"""Register multiple users with concurrent password hashing."""
for user in users:
if user.username in users_db:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Username {user.username} already exists",
)
# Hash all passwords concurrently — much faster than sequential
passwords = [user.password for user in users]
password_hashes = await hash_manager.hash_many_async(passwords)
registered = []
for user, password_hash in zip(users, password_hashes):
users_db[user.username] = {
"email": user.email,
"password_hash": password_hash,
}
registered.append(user.username)
return {
"message": f"Successfully registered {len(registered)} users",
"users": registered,
}
For bulk operations outside of a web request — such as a one-off data migration — you can combine hash_many_async with asyncio.gather across multiple batches to maximise throughput. Async batch processing is typically 3–5x faster than sequential hashing for 10 passwords, and 8–10x faster at 100 passwords.
Basic Async Example
If you’re not using a web framework, run async code directly with asyncio.run:
import asyncio
from hash_forge import HashManager
async def main():
manager = HashManager.from_algorithms("argon2")
# Non-blocking hash
hashed = await manager.hash_async("my_password")
print(f"Hashed: {hashed}")
# Non-blocking verify
is_valid = await manager.verify_async("my_password", hashed)
print(f"Valid: {is_valid}") # True
# Non-blocking rehash check
needs_rehash = await manager.needs_rehash_async(hashed)
print(f"Needs rehash: {needs_rehash}") # False
asyncio.run(main())
