Checking Database Permissions (MSSQL) with SQL Alchemy and Pydantic

Checking Database Permissions (MSSQL) with SQL Alchemy and Pydantic

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
        with engine.connect():

    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)
    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

        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

async def lifespan(app: FastAPI):
    Lifespan method for fastapi
    table_name = "tableName"
    schema_name = "schemaName"
        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)
        # Do something

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

← Previous
Add Outlook Shared Mailbox Rules for Disabling Notifications

Get rid of the all distractions when you are focusing.

Next →
Django Translation Using .PO File

This post is deprecated. Please follow the official documentation. When comes to using multiple …

Share Your Thoughts
M↓ Markdown