CSV Operations

Overview

The CSV Operations module provides three essential functions for managing persistent storage of license plate scan records. These functions handle data initialization, reading, and writing operations for the application’s CSV-based database.

Functions

ensure_csv()

Initializes the CSV database file with proper headers if it doesn’t already exist.

Function Signature

def ensure_csv() -> None

Parameters

This function takes no parameters.

Behavior

Checks if the CSV file exists at the configured DATA_PATH
If the file does not exist, creates it with the standard CSV header
If the file already exists, performs no action (idempotent operation)

Implementation

def ensure_csv():
    if not os.path.exists(DATA_PATH):
        with open(DATA_PATH, "w", newline="", encoding="utf-8") as f:
            csv.writer(f).writerow(CSV_HEADER)

The function writes the following header row:

CSV_HEADER = ["id", "fecha_hora", "matricula", "propietario", 
              "tipo_vehiculo", "observacion", "imagen"]

Usage Example

from app import ensure_csv

# Ensure database exists before any operations
ensure_csv()
print("CSV database initialized")

read_csv()

Reads all license plate records from the CSV database and returns them as a list of dictionaries.

Function Signature

def read_csv() -> list[dict[str, str]]

Parameters

This function takes no parameters.

Return Value

records

list[dict]

A list of dictionaries, where each dictionary represents one license plate record with the following keys:

string

Unique identifier for the record (sequential number starting from 1)

fecha_hora

string

Timestamp when the record was created (format: “YYYY-MM-DD HH:MM:SS”)

matricula

string

The extracted license plate text (uppercase, alphanumeric)

propietario

string

Vehicle owner name (optional, user-provided)

tipo_vehiculo

string

Type of vehicle (e.g., “auto”, “moto”, “camion”)

observacion

string

Additional notes or observations about the vehicle/scan

imagen

string

Filename of the uploaded image stored in the uploads directory

Implementation

def read_csv():
    ensure_csv()
    with open(DATA_PATH, "r", encoding="utf-8") as f:
        reader = list(csv.DictReader(f))
    return reader

The function automatically calls ensure_csv() before reading, guaranteeing that the CSV file exists even on first access.

Usage Example

from app import read_csv

# Retrieve all records
all_records = read_csv()

# Display record count
print(f"Total records in database: {len(all_records)}")

# Iterate through records
for record in all_records:
    print(f"ID: {record['id']}, Plate: {record['matricula']}, Date: {record['fecha_hora']}")

# Filter records by vehicle type
autos = [r for r in all_records if r['tipo_vehiculo'] == 'auto']
print(f"Found {len(autos)} auto records")

# Search by plate number
search_plate = "ABC123"
matches = [r for r in all_records if r['matricula'] == search_plate]
if matches:
    print(f"Found record for plate {search_plate}")

write_csv()

Writes a complete list of records to the CSV database, replacing all existing data.

Function Signature

def write_csv(rows: list[dict[str, str]]) -> None

Parameters

rows

list[dict]

required

A list of dictionaries representing license plate records to write. Each dictionary must contain all required CSV header fields:

id: Unique record identifier
fecha_hora: Timestamp string
matricula: License plate text
propietario: Owner name
tipo_vehiculo: Vehicle type
observacion: Additional notes
imagen: Image filename

Behavior

Opens the CSV file in write mode, replacing all existing content
Writes the standard CSV header row
Writes all provided rows to the file
Uses UTF-8 encoding to support international characters
Uses newline="" parameter to ensure proper CSV formatting across platforms

This function performs a complete replacement of the CSV file. To modify data, you must:

Read existing records with read_csv()
Modify the list of records
Write the modified list back with write_csv()

Implementation

def write_csv(rows):
    with open(DATA_PATH, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=CSV_HEADER)
        writer.writeheader()
        writer.writerows(rows)

Usage Examples

Adding a New Record

from app import read_csv, write_csv
from datetime import datetime

# Read existing records
records = read_csv()

# Create new record
new_record = {
    "id": str(len(records) + 1),
    "fecha_hora": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    "matricula": "XYZ789",
    "propietario": "Juan Pérez",
    "tipo_vehiculo": "auto",
    "observacion": "Vehículo de entrega",
    "imagen": "matricula_20240315_143022.jpg"
}

# Append and save
records.append(new_record)
write_csv(records)
print("New record added successfully")

Deleting a Record

from app import read_csv, write_csv
import os

# Read all records
records = read_csv()

# Delete record with specific ID
id_to_delete = "5"
filtered_records = [r for r in records if r["id"] != id_to_delete]

# Get deleted record info to remove image
deleted = [r for r in records if r["id"] == id_to_delete]
if deleted:
    img_path = os.path.join(UPLOADS_DIR, deleted[0]["imagen"])
    if os.path.exists(img_path):
        os.remove(img_path)

# Save updated records
write_csv(filtered_records)
print(f"Record {id_to_delete} deleted")

Updating a Record

from app import read_csv, write_csv

# Read existing records
records = read_csv()

# Find and update a specific record
for record in records:
    if record["id"] == "3":
        record["observacion"] = "Updated observation"
        record["propietario"] = "New Owner Name"
        break

# Save changes
write_csv(records)
print("Record updated successfully")

Bulk Operations

from app import read_csv, write_csv

# Read all records
records = read_csv()

# Normalize all plate numbers to uppercase
for record in records:
    record["matricula"] = record["matricula"].upper()

# Remove records with no detected plates
records = [r for r in records if r["matricula"] != "NO_DETECTADA"]

# Save bulk changes
write_csv(records)
print(f"Processed {len(records)} valid records")

Configuration

These functions use globally configured paths defined in app.py:

BASE_DIR = os.path.dirname(__file__)
DATA_PATH = os.path.join(BASE_DIR, "data", "registros.csv")
UPLOADS_DIR = os.path.join(BASE_DIR, "uploads")

CSV_HEADER = ["id", "fecha_hora", "matricula", "propietario", 
              "tipo_vehiculo", "observacion", "imagen"]

Directory Initialization

The application ensures required directories exist at startup:

os.makedirs(os.path.dirname(DATA_PATH), exist_ok=True)
os.makedirs(UPLOADS_DIR, exist_ok=True)

Application Integration

These CSV operations are used throughout the Flask application:

In the Save Route (`/guardar`)

@app.route("/guardar", methods=["POST"])
def guardar():
    ensure_csv()  # Initialize database
    
    # ... process upload and OCR ...
    
    rows = read_csv()  # Get existing records
    new_id = str(len(rows) + 1)  # Generate new ID
    
    rows.append({
        "id": new_id,
        "fecha_hora": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "matricula": matricula,
        "propietario": propietario,
        "tipo_vehiculo": tipo,
        "observacion": obs,
        "imagen": filename
    })
    
    write_csv(rows)  # Save updated records
    return redirect(url_for("index"))

In the Records Display Route (`/registros`)

@app.route("/registros")
def registros():
    data = read_csv()  # Load all records for display
    return render_template("registros.html", registros=data)

In the Delete Route (`/eliminar/<id>`)

@app.route("/eliminar/<id>")
def eliminar(id):
    rows = read_csv()  # Load current records
    filtered = [r for r in rows if r["id"] != id]  # Filter out deleted record
    
    # Delete associated image file
    deleted = [r for r in rows if r["id"] == id]
    if deleted:
        img_path = os.path.join(UPLOADS_DIR, deleted[0]["imagen"])
        if os.path.exists(img_path):
            os.remove(img_path)
    
    write_csv(filtered)  # Save updated records
    return redirect(url_for("registros"))

Error Handling

These functions do not include explicit error handling. File I/O errors (permission denied, disk full, etc.) will raise exceptions that should be caught by calling code.

Consider wrapping CSV operations in try-except blocks for production use:

try:
    records = read_csv()
    # ... process records ...
    write_csv(records)
except IOError as e:
    print(f"Error accessing CSV file: {e}")
    # Handle error appropriately
except csv.Error as e:
    print(f"Error parsing CSV data: {e}")
    # Handle error appropriately

Thread Safety

These functions are not thread-safe. Concurrent writes can result in data loss or corruption. For production deployments with multiple workers or concurrent requests, consider:

Implementing file locking mechanisms
Using a proper database (SQLite, PostgreSQL, etc.)
Implementing request queuing for write operations

Performance Considerations

Full File Read/Write: Every operation reads or writes the entire CSV file. For large datasets (>10,000 records), consider migrating to a database.
No Indexing: Searching records requires iterating through the entire list. Response time degrades linearly with dataset size.
Memory Usage: The entire dataset is loaded into memory. Monitor memory consumption for large deployments.

Migration Path

For applications outgrowing CSV storage, these functions provide a clean abstraction layer. To migrate to a database:

Keep the function signatures unchanged
Replace CSV file operations with database queries
Application code using read_csv() and write_csv() requires no changes

extract_plate_from_image() - Extracts license plate text before saving to CSV
Flask routes: /guardar, /registros, /eliminar/<id> - Primary consumers of these operations

Endpoints

Core Functions

Overview

Functions

ensure_csv()

Function Signature

Parameters

Behavior

Implementation

Usage Example

read_csv()

Function Signature

Parameters

Return Value

Implementation

Usage Example

write_csv()

Function Signature

Parameters

Behavior

Implementation

Usage Examples

Configuration

Directory Initialization

Application Integration

In the Save Route (`/guardar`)

In the Records Display Route (`/registros`)

In the Delete Route (`/eliminar/<id>`)

Error Handling

Thread Safety

Performance Considerations

Migration Path

Build docs developers (and LLMs) love

Endpoints

Core Functions

​Overview

​Functions

​ensure_csv()

​Function Signature

​Parameters

​Behavior

​Implementation

​Usage Example

​read_csv()

​Function Signature

​Parameters

​Return Value

​Implementation

​Usage Example

​write_csv()

​Function Signature

​Parameters

​Behavior

​Implementation

​Usage Examples

​Configuration

​Directory Initialization

​Application Integration

​In the Save Route (/guardar)

​In the Records Display Route (/registros)

​In the Delete Route (/eliminar/<id>)

​Error Handling

​Thread Safety

​Performance Considerations

​Migration Path

​Related Functions

Build docs developers (and LLMs) love

Overview

Functions

ensure_csv()

Function Signature

Parameters

Behavior

Implementation

Usage Example

read_csv()

Function Signature

Parameters

Return Value

Implementation

Usage Example

write_csv()

Function Signature

Parameters

Behavior

Implementation

Usage Examples

Configuration

Directory Initialization

Application Integration

In the Save Route (`/guardar`)

In the Records Display Route (`/registros`)

In the Delete Route (`/eliminar/<id>`)

Error Handling

Thread Safety

Performance Considerations

Migration Path

Related Functions