Using Peewee with two (or more) MySQL users to implement principle of least privilege

peewee_two_mysql_users_least_privilege.py

# Naive approach:
# 1. Create two MySQLDatabase instances for the same MySQL database:
# 	* One instantiated with your CRUD-privileged user
# 	* One instantiated with your DDL-privileged user
# 
# 2. Next, define your test model with `database=crud_instance` in
# 	its Meta subclass.
#
# 3. Then establish a database connection from the ddl_instance and
# 	issue `create_tables()`.
#
# 4. Finally, use the crud_instance to create, read, update, and delete
# 	throughout your application happily ever after.

# Error:
# CREATE command denied to user 'CRUD'@'localhost' for table 'test'

# Analysis:
# As soon as you call `create_tables()`, your test model is instantiated.
# Peewee will automatically try to create the corresponding tables, and it
# will use the database instance you've included in the model's Meta subclass
# (the crud_instance whose MySQL user doesn't have DDL privileges).

# Solution:
# Use Peewee's DatabaseProxy instance as a placeholder for your model's
# "database" property in its Meta subclass.


from peewee import Model, CharField, DatabaseProxy
from playhouse.mysql_ext import MySQLConnectorDatabase # Let's show some love for the offical MySQL driver

proxy_instance = DatabaseProxy()

class Base(Model):
	class Meta:
		database = proxy_instance

class Test(Base):
	test_column = CharField()

ddl_instance = MySQLConnectorDatabase(
	'my_database',
	host = 'localhost',
	user = 'DDL',
	password = 'secret')

crud_instance = MySQLConnectorDatabase(
	'my_database',
	host = 'localhost',
	user = 'CRUD',
	password = 'secret')

def issue_ddl():
	ddl_instance.connection()
	ddl_instance.create_tables([Test])
	ddl_instance.commit()

	if not ddl_instance.is_closed():
		ddl_instance.close()

def issue_crud():
	ddl_instance.connection()
	test_row = Test(test_column = "Inserted value")
	test_row.save()
	ddl_instance.commit()

proxy_instance.initialize(ddl_instance)
issue_ddl()

proxy_instance.initialize(crud_instance)
issue_crud()


# Output:
# mysql> show tables;
# +-----------------------+
# | Tables_in_my_database |
# +-----------------------+
# | test                  |
# +-----------------------+
# 1 row in set (0.00 sec)

# mysql> select * from test;
# +----+----------------+
# | id | test_column    |
# +----+----------------+
# |  1 | Inserted value |
# +----+----------------+
# 1 row in set (0.00 sec)