Pandas MySQL Insert: Append DataFrames and Verify Records
Insert Pandas DataFrames into MySQL with SQLAlchemy. Practise table copying, mark adjustments and validated CSV or Excel imports with a portable notebook.
Inserting DataFrame to MySQL database table by using to_sql() from Excel or CSV sources
Show Table of Contents ↓Create the original class-count DataFrame 🔝
The original synthetic class counts are Five=5, Six=2 and Three=3. These counts are a separate teaching dataset, not counts calculated from the student sample. All practice examples use fresh SQLite/SQLAlchemy databases and close their engines. Requires Pandas and SQLAlchemy.
import pandas as pd
import tempfile
from pathlib import Path
from io import StringIO
from sqlalchemy import create_engine, text, inspect
from sqlalchemy.engine import URL
from sqlalchemy.exc import IntegrityError
df = pd.DataFrame({"class": ["Five", "Six", "Three"], "No": [5, 2, 3]})
student_csv = 'id,name,class,mark,gender\n1,John Deo,Four,75,female\n2,Max Ruin,Three,85,male\n3,Arnold,Three,55,male\n4,Krish Star,Four,60,female\n5,John Mike,Four,60,female\n6,Alex John,Four,55,male\n7,My John Rob,Fifth,78,male\n8,Asruid,Five,85,male\n9,Tes Qry,Six,78,male\n10,Big John,Four,55,female\n11,Ronald,Six,89,female\n12,Recky,Six,94,female\n13,Kty,Seven,88,female\n14,Bigy,Seven,88,female\n15,Tade Row,Four,88,male\n16,Gimmy,Four,88,male\n17,Tumyu,Six,54,male\n18,Honny,Five,75,male\n19,Tinny,Nine,18,male\n20,Jackly,Nine,65,female\n21,Babby John,Four,69,female\n22,Reggid,Seven,55,female\n23,Herod,Eight,79,male\n24,Tiddy Now,Seven,78,male\n25,Giff Tow,Seven,88,male\n26,Crelea,Seven,79,male\n27,Big Nose,Three,81,female\n28,Rojj Base,Seven,86,female\n29,Tess Played,Seven,55,male\n30,Reppy Red,Six,79,female\n31,Marry Toeey,Four,88,male\n32,Binn Rott,Seven,90,female\n33,Kenn Rein,Six,96,female\n34,Gain Toe,Seven,69,male\n35,Rows Noump,Six,88,female\n'
students = pd.read_csv(StringIO(student_csv))
def student_engine():
engine = create_engine("sqlite:///:memory:")
with engine.begin() as connection:
students.to_sql("student", connection, index=False)
return engine
print(df)
assert df["No"].sum() == 10Expected output
class No
0 Five 5
1 Six 2
2 Three 3Configure a MySQL connection separately 🔝
Install SQLAlchemy and a compatible MySQL driver such as PyMySQL, then supply your private credentials. URL.create handles special characters in passwords. This runnable example creates a demonstration URL without connecting. See MySQL with SQLAlchemy and MySQL connection settings.
url = URL.create("mysql+pymysql", username="practice_user", password="practice_password",
host="localhost", database="practice_database", query={"charset": "utf8mb4"})
print(url.render_as_string(hide_password=True))
assert url.drivername == "mysql+pymysql"Expected output
mysql+pymysql://practice_user:***@localhost/practice_database?charset=utf8mb4Insert the DataFrame into a new table 🔝
Use name for the table, con for the engine/connection and index=False to omit disposable row labels. if_exists="append" creates a missing table or adds rows to an existing one; it does not match records or update them. Use distinct table names for class counts and student records because their schemas differ.
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
df.to_sql("class_counts", connection, if_exists="append", index=False)
with engine.connect() as connection:
saved = pd.read_sql(text('SELECT class, "No" FROM class_counts ORDER BY class'), connection)
print(saved)
assert saved["No"].sum() == 10 and len(saved) == 3
finally:
engine.dispose()Expected output
class No
0 Five 5
1 Six 2
2 Three 3Optional: insert into your MySQL practice database 🔝
This cell is disabled for Run All. To use it, install PyMySQL, configure the listed environment variables privately and enable run_mysql only for your practice database. It creates a new class-count table using fail, preventing accidental append on a second run. Verify the destination and privileges before enabling a write.
import os
run_mysql = False
if run_mysql:
private_url = URL.create("mysql+pymysql", username=os.environ["MYSQL_USER"],
password=os.environ["MYSQL_PASSWORD"], host=os.environ["MYSQL_HOST"],
database=os.environ["MYSQL_DATABASE"], query={"charset": "utf8mb4"})
engine = create_engine(private_url)
try:
with engine.begin() as connection:
df.to_sql("p2n_class_counts_practice", connection, if_exists="fail", index=False)
count = connection.execute(text("SELECT COUNT(*) FROM p2n_class_counts_practice")).scalar_one()
assert count == len(df)
print("MySQL practice insertion checked")
finally:
engine.dispose()
else:
print("Live MySQL insertion disabled; portable examples use SQLite.")Expected output
Live MySQL insertion disabled; portable examples use SQLite.Read one table and write another 🔝
Retain the original class Five table-copy workflow. Bind values in a query and write the selected DataFrame into a new student2 table. The embedded sample uses class Fifth for My John Rob; only Asruid and Honny have class Five. See MySQL read_sql() and the student sample.
engine = student_engine()
try:
with engine.begin() as connection:
selected = pd.read_sql(text("SELECT * FROM student WHERE class=:class ORDER BY id"),
connection, params={"class": "Five"})
selected.to_sql("student2", connection, if_exists="fail", index=False)
copied = pd.read_sql(text("SELECT * FROM student2 ORDER BY id"), connection)
print(copied)
pd.testing.assert_frame_equal(copied, selected)
assert copied["id"].tolist() == [8, 18]
finally:
engine.dispose()Expected output
id name class mark gender
0 8 Asruid Five 85 male
1 18 Honny Five 75 maleAdd five marks and replace a practice table 🔝
This preserves the original read-modify-write example on an isolated practice table. replace drops and recreates a table; constraints and indexes may be lost. It is suitable only when deliberately recreating that table. A row UPDATE is demonstrated next.
engine = student_engine()
try:
with engine.begin() as connection:
original = pd.read_sql(text("SELECT * FROM student WHERE class=:class ORDER BY id"),
connection, params={"class": "Five"})
original.to_sql("student2", connection, index=False)
revised = original.copy()
revised["mark"] = revised["mark"] + 5
with engine.begin() as connection:
revised.to_sql("student2", connection, if_exists="replace", index=False)
checked = pd.read_sql(text("SELECT * FROM student2 ORDER BY id"), connection)
print(checked)
assert checked["mark"].tolist() == [90, 80]
assert original["mark"].tolist() == [85, 75]
finally:
engine.dispose()Expected output
id name class mark gender
0 8 Asruid Five 90 male
1 18 Honny Five 80 maleUpdate rows while retaining their table 🔝
Execute parameterized UPDATE statements for existing rows rather than recreating the table. This keeps its primary-key schema. Validate marks against your own grading rules before applying adjustments.
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
connection.execute(text("CREATE TABLE marks (id INTEGER PRIMARY KEY, mark INTEGER NOT NULL)"))
connection.execute(text("INSERT INTO marks VALUES (:id,:mark)"),
[{"id": 8, "mark": 85}, {"id": 18, "mark": 75}])
connection.execute(text("UPDATE marks SET mark=mark+:bonus WHERE id=:id"),
[{"id": 8, "bonus": 5}, {"id": 18, "bonus": 5}])
with engine.connect() as connection:
checked = pd.read_sql(text("SELECT * FROM marks ORDER BY id"), connection)
print(checked)
assert checked["mark"].tolist() == [90, 80]
assert inspect(engine).get_pk_constraint("marks")["constrained_columns"] == ["id"]
finally:
engine.dispose()Expected output
id mark
0 8 90
1 18 80Import CSV and Excel records before inserting 🔝
Preserve both input workflows using embedded data and temporary files. Excel requires openpyxl. In your own workflow replace these generated paths with the selected files. Inspect head() and tail(), then validate fields before insertion. See read_csv() and read_excel().
with tempfile.TemporaryDirectory(dir=Path.cwd()) as folder:
csv_file = Path(folder) / "student.csv"
excel_file = Path(folder) / "student.xlsx"
csv_file.write_text(student_csv, encoding="utf-8")
students.to_excel(excel_file, index=False, engine="openpyxl")
engine = create_engine("sqlite:///:memory:")
try:
for table, imported in [("csv_students", pd.read_csv(csv_file)),
("excel_students", pd.read_excel(excel_file, engine="openpyxl"))]:
with engine.begin() as connection:
imported.to_sql(table, connection, index=False, if_exists="fail")
checked = pd.read_sql(text(f"SELECT * FROM {table} ORDER BY id"), connection)
pd.testing.assert_frame_equal(checked, students)
print(students.head(2))
print(students.tail(2))
print("CSV and Excel inserted rows:", len(students))
finally:
engine.dispose()Expected output
id name class mark gender
0 1 John Deo Four 75 female
1 2 Max Ruin Three 85 male
id name class mark gender
33 34 Gain Toe Seven 69 male
34 35 Rows Noump Six 88 female
CSV and Excel inserted rows: 35Keep invalid import rows for review 🔝
These synthetic incoming records require a nonblank ID and nonnegative numeric amount. Keep rejected values intact. A database append should start with a validated table matching the target schema.
raw = pd.DataFrame({"id": ["001", "002", "003", "004"],
"amount": ["5.25", "20", "pending", "-1"]})
clean = raw.copy()
clean["amount"] = pd.to_numeric(clean["amount"], errors="coerce")
valid = clean["id"].notna() & clean["id"].str.strip().ne("") & clean["amount"].ge(0).fillna(False)
accepted = clean.loc[valid].copy()
review = raw.loc[~valid].copy()
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
accepted.to_sql("accepted_orders", connection, index=False)
checked = pd.read_sql(text("SELECT * FROM accepted_orders ORDER BY id"), connection)
print(checked)
print("Review IDs:", review["id"].tolist())
assert checked["id"].tolist() == ["001", "002"]
assert checked["amount"].sum() == 25.25
assert len(accepted) + len(review) == len(raw)
finally:
engine.dispose()Expected output
id amount
0 001 5.25
1 002 20.00
Review IDs: ['003', '004']Reject repeated IDs and verify transaction state 🔝
append is not duplicate protection. Pre-create keys and constraints when IDs must be unique. For MySQL rollback behavior use an appropriate transactional engine such as InnoDB; the tested example here uses SQLite. Pandas may wrap database exceptions in pandas.errors.DatabaseError.
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
connection.execute(text("CREATE TABLE imports (id INTEGER PRIMARY KEY, name TEXT)"))
students.head(2)[["id", "name"]].to_sql("imports", connection, index=False, if_exists="append")
try:
with engine.begin() as connection:
students.head(1)[["id", "name"]].to_sql("imports", connection, index=False, if_exists="append")
except (IntegrityError, pd.errors.DatabaseError) as error:
assert isinstance(error, IntegrityError) or isinstance(error.__cause__, IntegrityError)
print("Duplicate import rejected")
else:
raise AssertionError("Expected duplicate rejection")
with engine.connect() as connection:
assert connection.execute(text("SELECT COUNT(*) FROM imports")).scalar_one() == 2
finally:
engine.dispose()Expected output
Duplicate import rejectedInsertion choices and common pitfalls 🔝
name and con are required. if_exists defaults to fail; append inserts, while replace recreates. index defaults to True; set False for disposable labels. dtype chooses SQL types, chunksize batches writes, and method may alter insertion strategy. A returned row count can be driver-dependent; verify with SELECT. Keep class-count and student tables separate. Do not concatenate user-provided identifiers into queries; choose table names from a fixed list. Use private credentials and close/dispose connections. MySQL DDL can commit implicitly, so do not assume table replacement will roll back like ordinary row changes. append is not an upsert; implement a database-specific upsert only when intended. For full options see MySQL to_sql(), SQLite to_sql() and read_sql_table().
Exercise: insert a high-mark selection 🔝
Select ID, name and mark for class Four students with marks at least 80. Save them into a new practice table without an index, then verify IDs 15, 16 and 31.
Exercise solution 🔝
Select and inspect records before insertion, then verify the stored rows.
answer = students.loc[students["class"].eq("Four") & students["mark"].ge(80), ["id", "name", "mark"]].copy()
engine = create_engine("sqlite:///:memory:")
try:
with engine.begin() as connection:
answer.to_sql("high_marks", connection, if_exists="fail", index=False)
checked = pd.read_sql(text("SELECT * FROM high_marks ORDER BY id"), connection)
print(checked)
assert checked["id"].tolist() == [15, 16, 31]
finally:
engine.dispose()Expected output
id name mark
0 15 Tade Row 88
1 16 Gimmy 88
2 31 Marry Toeey 88Practice in Google Colab 🔝
Open in Google Colab View on GitHub
Run examples in order, inspect database records and try the exercise. Save a copy to keep edits.
Reference: Pandas MySQL insert documentation.
Subhendu Mohapatra
Author & Instructor at plus2net
I write and maintain practical tutorials on Python, PHP, SQL, JavaScript, HTML, jQuery, and web development at plus2net. The tutorials focus on clear explanations, working examples, and code that readers can test and adapt while learning.
Connect:
Subscribe to our YouTube Channel here
This article is written by plus2net.com team.
https://www.plus2net.com
Python Video Tutorials