Checking Database Permissions (MSSQL) with SQL Alchemy and Pydantic
Jul 08, 2024 · 5 Min Read · 7 Likes · 0 Comment
When working with different teams and using a database from another team, it’s essential to ensure that your application has the necessary permissions to utilize the database functionalities.In this article, we explored how to use Python with SQL Alchemy to check permissions for an application in a database without creating an entry in the database. By following the steps outlined here, you can verify the database connection, check if the schema exists, and ensure that your application will be able to interact with the database effectively.
Additionally, combining this approach with FAST API (assuming your application is FastAPI) allows you to perform these checks before deploying your application, ensuring a smooth and error-free experience for end-users.
Create Database Engine
Lets say you have the credentials for the database, lets create the database engine first:
from sqlalchemy import create_engine
def get_engine(db_url: str, **connection_args: Dict[Any, Any]) -> Engine:
"""
Create and return a SQLAlchemy engine object.
connection_args: it can contain credential informations, timeout or other
parameters depends on what information needed to create the database connection
"""
return create_engine(db_url, connect_args=connection_args)
Check Database Connection
Using the engine from previous step, we can check the database connection by trying to connect to the database:
from sqlalchemy import Engine
from sqlalchemy.exc import InterfaceError, OperationalError, ProgrammingError
def check_connection_exists(engine: Engine) -> None:
"""
Checks if DB exists
"""
try:
with engine.connect():
pass
except (InterfaceError, OperationalError, ProgrammingError) as excinfo:
raise CustomException(excinfo) from excinfo
This code snippet provides a way to check if a database connection can be established using the provided SQLAlchemy Engine object, and raises a custom exception if there is an error during the connection attempt.
Check Schema Exists
If the database exists, good news for us. Now we take the next step of checking if the schema which we have access to exists or not:
from sqlalchemy import inspect
def check_schema_exists(engine: Engine, schema_name:str = "SchemaName") -> None:
""""
Checks if schema exists in the database
"""
iengine = inspect(engine)
if not iengine.has_schema(schema_name):
raise CustomException(f"Schema {schema_name} does not exist")
In summary, this code checks if a schema exists in a database using SQLAlchemy’s inspect function. If the schema does not exist, it raises a custom exception.
Check Table Exists
In this step, we would like to verify if the table exists or not:
def check_table_exists(engine: Engine, table:str = "table_name") -> None:
"""
Checks if table in the Database exists
"""
iengine = inspect(engine)
try:
iengine.has_table(table)
except NoSuchTableError as excinfo:
raise CustomException(excinfo) from excinfo
This provides a convenient way to check if a table exists in a database using SQLAlchemy. It encapsulates the necessary logic and handles any potential exceptions that may occur during the table existence check.
Verify if the Table Matches the Model
We are going to verify if the table matches the pydantic model:
def check_table_schema_matches_model(
engine: Engine,
table_name:str= "table_name",
schema:str = "SchemaName",
pydantic_model: BaseModel ) -> None:
"""
Checks if schema from the model matches in the database
"""
iengine = inspect(engine)
table = pydantic_model.__table__ # your Pydentic Model
try:
columns = [
c["name"] for c in
iengine.get_columns(table_name, schema)
]
difference = set(table.get_column_names()) - set(columns)
if difference:
raise ColumnDoesNotExist(', '.join(difference)) # custom exception
except NoSuchTableError as excinfo:
raise CustomException(excinfo) from excinfo
This function ensures that the schema defined in the Pydantic model matches the schema in the database table by comparing the column names. If there are any missing columns, it raises an exception to indicate the discrepancy.
Check permissions
Now is the hardest part, check if the role has permissions to modify the database:
from sqlalchemy.sql import text
def check_permissions_from_table(
engine: Engine,
table_name:str ='table_name',
schema_name:str = "SchemaName") -> None:
"""
Ran raw SQL to check if the permissions exist in Database
"""
required_permissions = ['SELECT', 'INSERT', 'UPDATE', 'DELETE']
command = f"""
SELECT *
FROM fn_my_permissions('{table_name}', '{schema_name}');
"""
with engine.connect() as con:
response = con.execute(text(command))
permissions = map(lambda x: x[2], response)
difference = set(required_permissions) - set(permissions)
if difference:
raise MissingPermissionsError(', '.join(difference)) # custom errors
In this function, it connects to a database using SQLAlchemy with the engine provided in first step, executes a SQL query to check the permissions for a specific table and schema, and raises a custom error if any of the required permissions are missing.
Bonus: Combine it with FastAPI
We can leverage FastAPIs lifecycle methods to check if the database connection is working as expected on every time the application starts:
# global variable
engine = get_db_engine(DB_URL)
# rest of the code
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Lifespan method for fastapi
"""
table_name = "tableName"
schema_name = "schemaName"
try:
check_connection_exists(engine)
check_schema_exists(engine, schema_name)
check_table_exists(engine, table_name)
check_permissions_from_table(engine, table_name, schema_name)
check_table_schema_matches_model(engine, table_name, schema_name)
except:
# Do something
yield
In Conclusion
In summary, we explored how to use Python with SQL Alchemy to check permissions for an application in an MSSQL Database. By verifying the database connection and schema existence, we can ensure that our application will function smoothly without invasive operations. Combining this approach with FastAPI allows for pre-deployment checks, ensuring a seamless experience for end-users.
Last updated: Jul 13, 2024
I won't spam you. Unsubscribe at any time.