Skip to main content

Overview

The ZKTeco Biometric Server is built on Flask and designed to provide REST API access to ZKTeco biometric devices. The architecture supports two distinct operational modes:

Single-Device Mode

Direct connection to one biometric device with simple configuration

Multi-Device Mode

Dynamic management of multiple devices with persistence and parallel operations

Flask Application Structure

Both server implementations use Flask as the web framework with a consistent structure:
source/servidor.py
from flask import Flask, request, Response
import logging, os, json, threading

app = Flask(__name__)

# Configuration from environment variables
API_HOST = os.getenv("API_HOST", "0.0.0.0")
API_PORT = int(os.getenv("API_PORT", "5000"))
CERT_FILE = os.getenv("CERT_FILE", "cert.pem")
KEY_FILE = os.getenv("KEY_FILE", "key.pem")

Response Format

All endpoints use a consistent JSON response format:
source/servidor.py
def _json(data, status=200):
    return Response(
        json.dumps(data, ensure_ascii=False, indent=2),
        status=status, mimetype="application/json"
    )
The ensure_ascii=False parameter allows Unicode characters in responses, which is important for international names and locations.

Threading Model and Concurrency

Single-Device Mode: Simple Lock

In single-device mode (servidor.py), a single threading lock protects all device operations:
source/servidor.py
lock = threading.Lock()

@app.route("/users", methods=["GET"])
def get_users():
    with lock:
        conn = None
        try:
            conn = conectar()
            conn.disable_device()
            users = conn.get_users()
            conn.enable_device()
            # Process users...
        finally:
            if conn:
                conn.disconnect()
Why this matters:
  • Only one request can communicate with the device at a time
  • Prevents concurrent access issues and device communication errors
  • Simple and sufficient for single-device scenarios

Multi-Device Mode: Per-Device Locks

The multi-device server (server.py) uses a more sophisticated locking strategy:
source/server.py
devices_lock = threading.Lock()  # Protects the DEVICES registry
_locks = {k: threading.Lock() for k in DEVICES}  # Per-device operation locks

def get_lock(device_id):
    """Returns (creating if needed) the lock for a device."""
    with devices_lock:
        if device_id not in _locks:
            _locks[device_id] = threading.Lock()
        return _locks[device_id]
Two-Level Locking Strategy:
1

Registry Lock (devices_lock)

Protects read/write operations on the DEVICES dictionary and devices.json file. Short-lived and only held during device registration/modification.
2

Per-Device Lock (get_lock)

Each device has its own lock, allowing parallel operations across different devices while preventing concurrent access to the same device.

Parallel Operations Example

The /attendance/all endpoint fetches attendance from all devices in parallel:
source/server.py
@app.route("/attendance/all", methods=["GET"])
def get_all_attendance():
    result = []
    errors = {}
    rlock = threading.Lock()  # Protects shared result list

    def fetch(device_id):
        with get_lock(device_id):  # Device-specific lock
            conn = None
            try:
                conn = conectar(device_id)
                registros = conn.get_attendance()
                # Process records...
                with rlock:  # Protect shared data
                    result.extend(parcial)
            except Exception as e:
                with rlock:
                    errors[device_id] = str(e)
            finally:
                if conn:
                    conn.disconnect()

    # Get snapshot of device IDs
    with devices_lock:
        device_ids = list(DEVICES.keys())

    # Launch parallel threads
    threads = [threading.Thread(target=fetch, args=(d,)) for d in device_ids]
    for t in threads: t.start()
    for t in threads: t.join()
    
    return _json({"success": True, "total": len(result), "data": result})
This architecture allows querying 10 devices in parallel instead of sequentially, dramatically reducing response time for multi-device deployments.

Device Connection Management

Connection Pattern

Both servers follow a strict connection lifecycle pattern:
# 1. Acquire lock
with get_lock(device_id):
    conn = None
    try:
        # 2. Connect to device
        conn = conectar(device_id)
        
        # 3. Disable device (prevents beeping, improves stability)
        conn.disable_device()
        
        # 4. Perform operations
        data = conn.get_users()
        
        # 5. Re-enable device
        conn.enable_device()
        
        return _json({"success": True, "data": data})
    except Exception as e:
        return _json({"success": False, "error": str(e)}), 500
    finally:
        # 6. Always disconnect, even on error
        if conn:
            try:
                conn.enable_device()  # Ensure device is enabled
            except:
                pass  # Ignore errors during cleanup
            conn.disconnect()
Critical: Always disconnect in the finally blockFailing to disconnect properly can leave the device in a locked state, requiring a manual reboot.

Connection Factory Functions

Single-Device Mode:
source/servidor.py
def conectar():
    zk = ZK(ZK_IP, port=ZK_PORT, timeout=ZK_TIMEOUT,
            password=ZK_PASSWORD, force_udp=False, ommit_ping=False)
    return zk.connect()
Multi-Device Mode:
source/server.py
def conectar(device_id):
    cfg = DEVICES[device_id]
    zk = ZK(cfg["ip"], port=cfg["port"], timeout=cfg["timeout"],
            password=cfg["password"], force_udp=False, ommit_ping=False)
    return zk.connect()

Operating Modes Comparison

Use Case: Simple deployments with one biometric deviceConfiguration:
  • Device IP/port set via environment variables
  • No persistence layer required
  • Direct connection on every request
Advantages:
  • Simple setup and configuration
  • Minimal resource overhead
  • Easy to understand and debug
Limitations:
  • Cannot manage multiple devices
  • Configuration changes require server restart
  • No device registry or metadata storage
Use Case: Enterprise deployments with multiple biometric devicesConfiguration:
  • Devices registered via REST API
  • Persistent storage in devices.json
  • Each device can have unique settings
Advantages:
  • Dynamic device registration without restart
  • Parallel operations across devices
  • Device metadata and lifecycle tracking
  • Centralized management of multiple locations
Complexity:
  • More sophisticated locking mechanism
  • Persistent state management
  • Additional CRUD endpoints for device management

Request Lifecycle

Here’s how a typical request flows through the multi-device server:

Error Handling Strategy

The server implements comprehensive error handling:
try:
    conn = conectar(device_id)
except Exception as e:
    log.error(f"[{device_id}] Connection failed: {e}")
    return _json({
        "success": False, 
        "error": str(e),
        "ip": DEVICES[device_id]["ip"]
    }), 503

Logging Configuration

Both servers use Python’s standard logging with consistent formatting:
source/server.py
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
log = logging.getLogger(__name__)
Log Output Example:
2026-03-06 14:32:15 [INFO] Dispositivos cargados: ['principal', 'bodega']
2026-03-06 14:32:47 [INFO] [principal] Probando conexion...
2026-03-06 14:32:48 [INFO] Dispositivo registrado: [bodega] Bodega Principal → 192.168.1.207
Next Steps:

Build docs developers (and LLMs) love