-- schema.sql CREATE TABLE roles ( id SERIAL PRIMARY KEY, role_name TEXT );
CREATE TABLE users ( id SERIAL PRIMARY KEY, username TEXT, role_id INT REFERENCES roles(id) );
CREATE TABLE permissions ( id SERIAL PRIMARY KEY, action TEXT );
CREATE TABLE role_permissions ( role_id INT REFERENCES roles(id), permission_id INT REFERENCES permissions(id), PRIMARY KEY(role_id, permission_id) );
INSERT INTO roles(role_name) VALUES ('admin'), ('user'); INSERT INTO permissions(action) VALUES ('read'), ('write'), ('delete');
import psycopg2
DB_PARAMS = { "dbname": "rbac_db", "user": "postgres", "password": "password", "host": "localhost", "port": 5432 }
def check_permission(username, action): conn = psycopg2.connect(**DB_PARAMS) cur = conn.cursor() query = """ SELECT 1 FROM users u JOIN role_permissions rp ON u.role_id = rp.role_id JOIN permissions p ON rp.permission_id = p.id WHERE u.username = %s AND p.action = %s; """ cur.execute(query, (username, action)) allowed = cur.fetchone() print(f"Permission check for {username} and {action}: {'ALLOWED' if allowed else 'DENIED'}") cur.close() conn.close()
if name == "main": check_permission("alice", "read") check_permission("bob", "delete")