maxsql

MySQL in one line per operation.

maxsql is a small helper around mysql-connector-python. You save your connection once, and after that every operation (create, insert, select, update, delete, export, alter, import) is a single call. Connections are pooled, failed connections are retried with backoff, and errors are printed instead of crashing your script.

Table of contents

1. Features
2. Installation
3. Quick start
4. Connections
   • configure
   • default
   • use
   • configs
   • forget
   • Working with several databases at once
   • Hardcoding credentials for one run
5. Query parameters
6. Operations
   • create
   • insert
   • select
   • update
   • delete
   • export
   • alter
   • import_
   • cursor
7. Messages and the display flag
8. Settings
9. Retry and pooling behaviour
10. Where connections are saved
11. Full example script
12. Troubleshooting
13. Security notes

Features

• Save a connection once, use it from every script.
• One line per operation: create, insert, select, update, delete, export, alter, import_.
• Connection pooling, so each call does not open a fresh connection.
• Automatic retry with exponential backoff for transient connection errors.
• Several saved connections, with a default and a per-script override.
• Works with several databases in the same script through Database objects.
• numpy and pandas values (NaN, NaT, pd.NA, numpy integers, Timestamps) are converted to MySQL-safe values for you.
• import_ builds a table straight from a pandas DataFrame with type detection.
• alter adds missing columns by looking at sample values.
• Errors and warnings always print and the call returns None or False, so a failure never crashes your script.

Installation

pip install maxsql

This also installs mysql-connector-python and pandas for you.

Quick start

import maxsql

# Once per connection. Asks for host, user, password and database.
maxsql.configure()

# Create a table
maxsql.create("""
    CREATE TABLE IF NOT EXISTS customers (
        id INT AUTO_INCREMENT PRIMARY KEY,
        name VARCHAR(100),
        city VARCHAR(80),
        credit_limit INT
    )
""")

# Insert
new_id = maxsql.insert(
    "INSERT INTO customers (name, city, credit_limit) VALUES (%s, %s, %s)",
    ("Acme Ltd", "Lagos", 5000),
)

# Select
rows = maxsql.select("SELECT * FROM customers WHERE credit_limit > %s", (1000,))
print(rows)

# Update
maxsql.update("UPDATE customers SET credit_limit = %s WHERE id = %s", (7500, new_id))

# Delete
maxsql.delete("DELETE FROM customers WHERE id = %s", (new_id,))

After the first run you can remove the configure() line. The connection is saved on disk and picked up automatically.

Connections

A connection is saved under a name. By default the name is the database name, so a database called shop is saved as shop.

If you only ever save one connection, you never need default() or use(). They matter when you have several.

configure

Saves a connection. Run it once per connection.

maxsql.configure(
    host=None,        # asks if not given; press Enter to accept 127.0.0.1
    user=None,        # asks if not given
    password=None,    # asks if not given (input is hidden)
    database=None,    # asks if not given
    port=3306,
    local=False,      # True skips the host question and assumes 127.0.0.1
    name=None,        # connection name, defaults to the database name
    display=True,     # show the "Saved" confirmation
    default=False,    # True makes this the default connection
)

Returns True when saved and False when not.

Interactive, asks for everything:

maxsql.configure()

Local server, skip the host question:

maxsql.configure(local=True)

Fully non-interactive:

maxsql.configure(
    host="127.0.0.1",
    user="root",
    password="secret",
    database="shop",
)

Same database name on two different servers. Give each one its own name:

maxsql.configure(name="shop_local")
maxsql.configure(name="shop_prod")

If you try to save a name that already exists for a different server, maxsql refuses and prints a message showing both entries and how to keep both. Saving the same name for the same server replaces the saved details, which is how you update a password:

maxsql.configure(user="root", password="new_password", database="shop", local=True)

Make a connection the default while saving it:

maxsql.configure(name="shop_prod", default=True)

The first connection you ever save becomes the default automatically. After configure() runs, that connection is also the active one for the current script.

default

Sets, checks or turns off your default connection. The default is saved on disk and used by every script that does not call use().

maxsql.default("shop")        # make shop the default, returns True/False
maxsql.default()              # print and return the current default (or None)
maxsql.default(off=True)      # no default; with several saved, use() is then required
maxsql.default("shop", display=False)   # no confirmation message

Passing both a name and off=True raises a ValueError.

use

Chooses the connection for the current script only. It overrides the default without changing it.

maxsql.use("shop_prod")
rows = maxsql.select("SELECT COUNT(*) FROM customers")

Returns True when the connection exists and False (with a message) when it does not.

If several connections are saved, no default is set and you do not call use(), maxsql prints which connections exist and how to choose one.

configs

Lists every available connection. Passwords are hidden. The active one is marked, the default is labelled, and in-memory connections are labelled "this run only".

maxsql.configs()

Example output:

▶ shop         root@127.0.0.1:3306/shop   password: ********   (default)
  shop_prod    app@10.0.0.5:3306/shop     password: ********

forget

Removes a saved connection. If it was the default, the default is cleared.

maxsql.forget("shop_prod")
maxsql.forget("shop_prod", display=False)   # silent

Returns True when removed and False when nothing with that name exists.

Working with several databases at once

use() switches the whole script. To talk to two databases at the same time, create a Database for each one. A Database("name") is tied to that connection no matter what use() says.

import maxsql

shop = maxsql.Database("shop")
accounting = maxsql.Database("accounting")

customers = shop.select("SELECT * FROM customers")
accounting.insert(
    "INSERT INTO invoices (customer, amount) VALUES (%s, %s)",
    ("Acme Ltd", 249.95),
)

Every module-level function has a method with the same name and arguments on Database: create, insert, select, update, delete, export, alter, import_, cursor.

Database() with no name follows whatever use() chose, which is exactly what maxsql.select(...) does.

Hardcoding credentials for one run

Pass the details to use() and nothing is saved to disk. This is handy for scripts, containers and CI.

maxsql.use("shop", host="127.0.0.1", user="root", password="secret")

Keyword details override any saved details with the same name, for this run only. host defaults to 127.0.0.1, port to 3306 and database to the connection name. A user and password are required.

Reading credentials from environment variables is the safest way to hardcode:

import os
import maxsql

maxsql.use(
    "shop",
    host=os.environ["DB_HOST"],
    user=os.environ["DB_USER"],
    password=os.environ["DB_PASSWORD"],
)

Query parameters

Every query function accepts parameters in four shapes:

# 1. No parameters
maxsql.select("SELECT * FROM customers")

# 2. A single value
maxsql.select("SELECT * FROM customers WHERE id = %s", 7)

# 3. A tuple or list, with %s placeholders
maxsql.select("SELECT * FROM customers WHERE credit_limit > %s AND city = %s", (1000, "Lagos"))

# 4. A dict, with %(name)s placeholders
maxsql.select(
    "SELECT * FROM customers WHERE credit_limit > %(limit)s AND city = %(city)s",
    {"limit": 1000, "city": "Lagos"},
)

Always pass values as parameters instead of building the query with f-strings. It prevents SQL injection and handles quoting for you.

Values are cleaned before they are sent: numpy and pandas scalars become plain Python values, and NaN, NaT, pd.NA and infinity become None (SQL NULL).

import numpy as np
import pandas as pd

maxsql.insert(
    "INSERT INTO daily_sales (order_count, avg_order_value, report_date) VALUES (%s, %s, %s)",
    (np.int64(38), np.float64("nan"), pd.Timestamp("2026-08-21")),
)
# the NaN is stored as NULL

Table and column names cannot be passed as parameters. Only alter and import_ validate names for you (letters, numbers and underscores, up to 64 characters). In your own raw queries, never insert user-supplied names directly.

Operations

create

Runs a statement such as CREATE TABLE and commits.

maxsql.create("""
    CREATE TABLE IF NOT EXISTS orders (
        id INT AUTO_INCREMENT PRIMARY KEY,
        customer VARCHAR(80),
        product VARCHAR(80),
        ordered_at DATETIME,
        quantity INT,
        amount DECIMAL(10, 2)
    )
""")

Returns True on success and False on failure. Pass display=False to hide the success message:

ok = maxsql.create("CREATE TABLE IF NOT EXISTS t (id INT)", display=False)

insert

Inserts a row and returns the new row id (None on error).

row_id = maxsql.insert(
    "INSERT INTO orders (customer, product, ordered_at) VALUES (%s, %s, %s)",
    ("Acme Ltd", "Widget A", "2026-10-18 16:30:00"),
)
print(row_id)

Inserting in a loop:

new_orders = [
    ("Globex", "Widget B", "2026-10-19 12:30:00"),
    ("Initech", "Widget C", "2026-10-19 15:00:00"),
]
for customer, product, ordered_at in new_orders:
    maxsql.insert(
        "INSERT INTO orders (customer, product, ordered_at) VALUES (%s, %s, %s)",
        (customer, product, ordered_at),
    )

insert is silent by default because it usually runs in loops. Turn the confirmation on with display=True:

maxsql.insert("INSERT INTO t (id) VALUES (%s)", (1,), display=True)

For thousands of rows, use import_ with a DataFrame instead of a loop.

select

Returns a list of rows (tuples). Returns None on error.

rows = maxsql.select("SELECT customer, product FROM orders WHERE customer = %s", ("Acme Ltd",))
for customer, product in rows:
    print(customer, "ordered", product)

With column names. Pass headers=True to get (rows, column_names):

rows, columns = maxsql.select("SELECT * FROM orders", headers=True)
print(columns)    # ['id', 'customer', 'product', 'ordered_at', 'quantity', 'amount']

Straight into a DataFrame:

import pandas as pd

rows, columns = maxsql.select("SELECT * FROM orders", headers=True)
df = pd.DataFrame(rows, columns=columns)

Single value:

total = maxsql.select("SELECT COUNT(*) FROM orders")[0][0]

Check for None before indexing if the query might fail.

update

Returns the number of rows affected (None on error).

changed = maxsql.update(
    "UPDATE orders SET quantity = %s, amount = %s WHERE id = %s",
    (5, 249.95, 1),
)
print(changed, "row(s) updated")

With a confirmation message:

maxsql.update("UPDATE orders SET quantity = 0", (), display=True)

delete

Returns the number of rows deleted (None on error).

removed = maxsql.delete("DELETE FROM orders WHERE id = %s", (1,))
print(removed, "row(s) deleted")

export

Runs one or more queries and saves each result as a CSV file. Needs pandas.

files = maxsql.export(
    [
        ["Orders", "SELECT * FROM orders"],
        ["AcmeOrders", "SELECT * FROM orders WHERE customer = 'Acme Ltd'"],
    ],
    output_dir="TABLES",
)
print(files)    # ['TABLES/Orders.csv', 'TABLES/AcmeOrders.csv']

• queries is a list of [name, query] pairs.
• output_dir defaults to "TABLES" and is created if it does not exist.
• Names must be plain file names with no folders. A name like "../x" is skipped with a warning.
• A query that fails is skipped and the others still run.
• Returns the list of files written.

Silent version:

maxsql.export([["Customers", "SELECT * FROM customers"]], output_dir="out", display=False)

alter

Adds any columns that do not exist yet, picking each type from a sample value. It never changes or removes existing columns.

added = maxsql.alter(
    "orders",
    {
        "sales_rep": "Jane Doe",         # VARCHAR(200)
        "items": 12,                     # INT
        "tax_rate": 7.5,                 # DOUBLE
        "paid": False,                   # BOOLEAN
        "invoice_ref": "INV-2026-0042",  # VARCHAR(200), a plain string stays a string
        "notes": None,                   # DOUBLE, because None defaults to a number
    },
)
print(added)    # list of columns that were added

Type inference from the sample value:

Sample value	MySQL type
bool	BOOLEAN
int up to 2,147,483,647	INT
larger int	BIGINT
float	DOUBLE
datetime.datetime	DATETIME
datetime.date	DATE
str	VARCHAR(200)
None	DOUBLE, or VARCHAR(200) if the column is in string_keys

Making a None sample a text column:

maxsql.alter("orders", {"notes": None}, string_keys=("notes",))

Real dates:

import datetime

maxsql.alter("orders", {
    "created_at": datetime.datetime.now(),   # DATETIME
    "invoice_day": datetime.date.today(),    # DATE
})

Column name matching is case-insensitive, like MySQL. Table and column names are validated (letters, numbers, underscores, 64 characters max) and a bad name raises a ValueError. alter returns an empty list if the table does not exist.

import_

Creates a table from a pandas DataFrame (if it does not exist) and bulk-inserts every row. The name ends with an underscore because import is a Python keyword.

import pandas as pd
import maxsql

df = pd.DataFrame({
    "product": ["Widget A", "Widget B", "Widget C"],
    "units_sold": [240, 180, 95],
    "revenue": [4800.0, 5400.0, 3325.5],
    "last_sale": ["2026-10-04", "2026-10-05", "2026-10-05"],
})

inserted = maxsql.import_(df, "sales_summary")
print(inserted, "rows inserted")

What it does:

• Column types come from the DataFrame dtypes.
  • Integers become INT, or BIGINT when values are large (BIGINT UNSIGNED beyond the signed range).
  • Floats become DOUBLE, so no precision is lost.
  • Booleans become BOOLEAN, timedeltas become TIME.
  • Datetimes become DATE when every time is midnight, otherwise DATETIME.
  • Text becomes VARCHAR(200), or TEXT, MEDIUMTEXT or LONGTEXT when the data is longer.
• Text columns that actually hold dates (2026-10-04, 04/10/2026, with or without a time) are detected and stored as real DATE or DATETIME values. Plain numbers such as 20261004 or bit strings such as 10000000 are never mistaken for dates.
• NaN, NaT and pd.NA are stored as NULL. Infinite values cannot be stored in MySQL, so they are saved as NULL and a warning tells you how many.
• Rows are sent in batches of 2,000 inside one transaction. If anything fails, the whole import is rolled back.
• Returns the number of rows inserted, or 0 on failure.

Notes:

• If the table already exists, CREATE TABLE IF NOT EXISTS leaves it alone and the rows are inserted into it. The DataFrame columns must match the existing table.
• Column names must use only letters, numbers and underscores (64 characters max). Rename them first if needed:

df.columns = [c.strip().lower().replace(" ", "_") for c in df.columns]

• Passing something that is not a DataFrame raises a TypeError. An empty DataFrame prints a warning and returns 0.
• Silent version: maxsql.import_(df, "sales_summary", display=False).

Adding new columns to a table before importing a DataFrame with extra columns:

sample = df.iloc[0].to_dict()
maxsql.alter("sales_summary", sample)
maxsql.import_(df, "sales_summary")

cursor

Gives you the raw cursor and connection for anything the helpers do not cover.

cur, conn = maxsql.cursor()
if cur is not None:
    try:
        cur.execute("SELECT id, customer FROM orders WHERE id > %s", (10,))
        for row in cur.fetchall():
            print(row)
    finally:
        cur.close()
        conn.close()     # with pooling on, this hands the connection back to the pool

Always close the connection when you are done. With pooling, an unreleased connection stays checked out and can eventually leave no free connections. When something goes wrong, cursor() returns (None, None).

For several statements in one transaction:

cur, conn = maxsql.cursor()
try:
    cur.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", (50, 1))
    cur.execute("UPDATE accounts SET balance = balance + %s WHERE id = %s", (50, 2))
    conn.commit()
except Exception:
    conn.rollback()
    raise
finally:
    cur.close()
    conn.close()

Messages and the display flag

Errors and warnings always print and never raise (for example 🚫 Error: ... and ⚠️ ...). The call returns None or False so your script keeps running. Check the return value when it matters:

result = maxsql.select("SELECT * FROM missing_table")
if result is None:
    print("query failed")

display controls only the success messages. It exists on create, insert, update, delete, alter, export, import_, configure, default and forget.

Function	Default
create, alter, export, import_, configure, default, forget	True
insert, update, delete	False (they usually run in loops)
select	no display, it returns rows and has nothing to confirm

Settings

These are module-level constants in maxsql/maxsql.py. Change them before your first query.

Setting	Default	Meaning
USE_POOLING	True	Reuse connections instead of opening one per call
POOL_SIZE	5	Connections per pool
POOL_WAIT_TIMEOUT	30	Seconds a call waits for a free pooled connection
CONNECT_TIMEOUT	10	Seconds to wait when opening a connection
IMPORT_BATCH	2000	Rows per statement in import_
CONNECT_RETRY_ATTEMPTS	5	Attempts when opening a connection
CONNECT_RETRY_BASE_DELAY	2	First retry delay in seconds, doubling each time
CONNECT_RETRY_MAX_DELAY	30	Cap on the retry delay
CONFIG_FILE	~/.maxsql.json	Where saved connections live
DEFAULT_HOST	127.0.0.1	Host used when none is given

Because the code lives in the maxsql.maxsql submodule, set them there:

import maxsql

maxsql.maxsql.POOL_SIZE = 10
maxsql.maxsql.CONNECT_RETRY_ATTEMPTS = 3
maxsql.maxsql.USE_POOLING = False

Setting maxsql.POOL_SIZE = 10 on the package has no effect, because the functions read the value from the submodule. POOL_SIZE only applies to pools created after you change it, so set it before your first query.

Retry and pooling behaviour

Pooling. With USE_POOLING = True, each distinct set of connection details gets its own pool, shared by every Database object. Calls borrow a connection and hand it back. If every pooled connection is busy, the call waits up to POOL_WAIT_TIMEOUT seconds before giving up with a message.

Retry. When opening a connection fails, maxsql retries only errors that are likely to be temporary:

Error code	Meaning
2003	Cannot connect to MySQL server
2002	Cannot connect to local MySQL server
2013	Lost connection during the query
1040	Too many connections
1053	Server shutdown in progress

The wait doubles each time (2, 4, 8, 16 seconds) up to CONNECT_RETRY_MAX_DELAY. Errors that would never succeed on retry, such as a wrong password or an unknown database, fail immediately with a clear message. This helps when several processes start at once and hit the server together.

Where connections are saved

Saved connections live in one JSON file in your home folder: ~/.maxsql.json.

{
  "default": "shop",
  "profiles": {
    "shop": {
      "host": "127.0.0.1",
      "user": "root",
      "password": "secret",
      "database": "shop",
      "port": 3306
    }
  }
}

• The file is read from disk once and cached, so queries never touch it.
• It is written atomically (a temp file renamed over the real one), so a crash cannot leave a half-written file.
• A lock file keeps two processes from overwriting each other when both call configure().
• On Linux and macOS the file is created with permissions 0600, so only you can read it.
• An older ~/.mysql_helper.json file is read once if the new file does not exist yet, and older formats are converted automatically.

Full example script

A complete script that uses most of the library:

import datetime
import pandas as pd
import maxsql

# One-time setup (safe to remove after the first run)
maxsql.configure(local=True)

# Table
maxsql.create("""
    CREATE TABLE IF NOT EXISTS orders (
        id INT AUTO_INCREMENT PRIMARY KEY,
        customer VARCHAR(80),
        product VARCHAR(80),
        ordered_at DATETIME,
        quantity INT,
        amount DECIMAL(10, 2)
    )
""", display=False)

# Insert
order_id = maxsql.insert(
    "INSERT INTO orders (customer, product, ordered_at) VALUES (%s, %s, %s)",
    ("Acme Ltd", "Widget A", datetime.datetime(2026, 10, 18, 16, 30)),
)

# Add columns we forgot
maxsql.alter("orders", {"sales_rep": "Jane Doe", "tax_rate": 7.5})

# Update with the final quantity and amount
maxsql.update(
    "UPDATE orders SET quantity = %s, amount = %s, sales_rep = %s WHERE id = %s",
    (5, 249.95, "Jane Doe", order_id),
)

# Read into a DataFrame
rows, columns = maxsql.select("SELECT * FROM orders", headers=True)
df = pd.DataFrame(rows, columns=columns)
print(df)

# Bulk-load a DataFrame into a new table
summary = pd.DataFrame({"product": ["Widget A", "Widget B"], "revenue": [4800.0, 5400.0]})
maxsql.import_(summary, "sales_summary")

# Export results to CSV
maxsql.export([
    ["Orders", "SELECT * FROM orders"],
    ["TopProducts", "SELECT * FROM sales_summary ORDER BY revenue DESC"],
])

# Clean up
maxsql.delete("DELETE FROM orders WHERE id = %s", (order_id,))

Troubleshooting

"No saved connection." Run maxsql.configure() once, or call maxsql.use("name", host=..., user=..., password=...) in your script.

"Several connections are saved ..." You have more than one saved connection and no default. Choose one for the script with maxsql.use("name"), or set a default once with maxsql.default("name").

"No saved connection named 'x'." The name is wrong. Run maxsql.configs() to see what exists.

"Not saved: 'x' is already saved for a different server." The same database name is already saved for another host or port. Save it under its own name, for example maxsql.configure(name="x_prod"), or run maxsql.forget("x") first to replace it.

"Error connecting to MySQL: Access denied." The user or password is wrong. Run maxsql.configure() again with the same name to replace the saved details.

"Error connecting to MySQL: Unknown database." The database does not exist on that server. Create it first, or check the name.

"No free connection after waiting 30s." Every pooled connection is in use. You probably opened connections with cursor() and did not close them. Call conn.close() when finished, or raise maxsql.maxsql.POOL_SIZE.

ImportError: maxsql needs the MySQL driver. The MySQL driver is missing. This can happen if you copied the files by hand instead of using pip install maxsql. Install it manually:

pip install mysql-connector-python

ImportError: This function needs pandas. pandas is missing, and export and import_ need it. Install it manually:

pip install pandas

ValueError: Invalid table name ... alter and import_ only accept names made of letters, numbers and underscores, up to 64 characters.

TypeError: import_() needs a pandas DataFrame Convert your data first, for example pd.DataFrame(data).

Security notes

• Passwords are stored in plain text in ~/.maxsql.json. The file is private to your user on Linux and macOS, but on Windows the permission bits are mostly ignored, so protect your home folder accordingly.
• For shared machines, servers and CI, prefer use() with environment variables so no password is written to disk.
• Never commit ~/.maxsql.json or scripts with hardcoded passwords to a public repository.
• Use query parameters for all values. Only alter and import_ validate table and column names, so never put user input into names in raw queries.
• Give the MySQL user only the privileges your script needs.