DEV Community

Cover image for Sitri = Vault+Pydantic: configuration with clear structure and validation
Alexander Lavrov
Alexander Lavrov

Posted on • Updated on

Sitri = Vault+Pydantic: configuration with clear structure and validation

Introduction

In this article, I will talk about the configuration for your services using the Vault (KV and so far only the first version, i.e. without versioning secrets) and Pydantic (Settings) under the patronage of Sitri.

So, let's say that we have a superapp service with configs set up in Vault and authentication using AppRole, we'll set it up like this (I'll leave the policies setting for access to secret engines and the secrets themselves behind the scenes, since this is quite simple and the article is not about this):

Key                        Value
---                        -----
bind_secret_id             true
local_secret_ids           false
policies                   [superapp_service]
secret_id_bound_cidrs      <nil>
secret_id_num_uses         0
secret_id_ttl              0s
token_bound_cidrs          []
token_explicit_max_ttl     0s
token_max_ttl              30m
token_no_default_policy    false
token_num_uses             50
token_period               0s
token_policies             [superapp_service]
token_ttl                  20m
token_type                 default
Enter fullscreen mode Exit fullscreen mode

Note: naturally, if you have the opportunity and the application goes into production mode, then secret_id_ttl is better to do not infinite, setting 0 seconds.

Superapp requires configuration of database connection, the connection to kafka and faust configuration for cluster of workers.

At the end, we should have this structure of our test-project:

super_app
├── config
│ ├── __init__.py
│ ├── provider_config.py
│ ├── faust_settings.py
│ ├── app_settings.py
│ ├── kafka_settings.py
│ └── database_settings.py
├── __init__.py
└── main.py
Enter fullscreen mode Exit fullscreen mode

Bake Sitri provider

Basic library documentation has a simple example, configuration via the vault provider, however, it does not cover all the features and can be useful if your application is configured easily enough.

So, first of all, let's configure the vault provider in a file provider_config.py:

import hvac  

from sitri.providers.contrib.vault import VaultKVConfigProvider  
from sitri.providers.contrib.system import SystemConfigProvider  

configurator = SystemConfigProvider(prefix="superapp")  
ENV = configurator.get("env")  


def vault_client_factory() -> hvac.Client:  
    client = hvac.Client(url=configurator.get("vault_api"))  

    client.auth_approle(  
        role_id=configurator.get("role_id"),  
  secret_id=configurator.get("secret_id"),  
  )  

    return client  


provider = VaultKVConfigProvider(  
    vault_connector=vault_client_factory, mount_point=f"{configurator.get('app_name')}/{ENV}"  
)
Enter fullscreen mode Exit fullscreen mode

In this code, we get variables from the environment using the system provider to configure the connection to vault, i.e. the following variables must be exported initially:

export SUPERAPP_ENV=dev
export SUPERAPP_APP_NAME=superapp
export SUPERAPP_VAULT_API=https://your-vault-host.domain
export SUPERAPP_ROLE_ID=<YOUR_ROLE_ID>
export SUPERAPP_SECRET_ID=<YOUR_SECRET_ID>
Enter fullscreen mode Exit fullscreen mode

The example assumes that the base mount_point to your secrets for a specific environment will contain the application name and the environment name, which is why we exported SUPERAPP_ENV. We will define the path to the secrets of individual parts of the application in the settings classes later, so we leave it empty in the secret_path vault provider argument.

Settings classes

DBSetting - database connection

To create the settings class, we must use VaultKVSettings as the base class.

File database_settings.py:

from pydantic import Field  

from sitri.settings.contrib.vault import VaultKVSettings  

from superapp.config.provider_config import provider  


class DBSettings(VaultKVSettings):  
    user: str = Field(..., vault_secret_key="username")  
    password: str = Field(...)  
    host: str = Field(...)  
    port: int = Field(...)  

    class Config:  
        provider = provider  
        default_secret_path = "db"
Enter fullscreen mode Exit fullscreen mode

As you can see, config data for the database connection is quite simple. This class will by default look at the secret superapp/dev/db, as we specified in the Config class. At first glance these are simple pydantic.Field, but they all have an extra argument vault_secret_key - it is needed when the key in the secret does not match the name (alias) of the pydantic field in our class, if vault_secret_key is not specified, the provider will search for the key by the field alias.

For example, in our superapp, it is assumed that the superapp/dev/db secret has the "password" and "username" keys, but we want the latter to be placed in the "user" field for convenience and brevity.

Let's put the following data in the above secret:

{
  "host": "testhost",
  "password": "testpassword",
  "port": "1234",
  "username": "testuser"
}
Enter fullscreen mode Exit fullscreen mode

Now, if we run this code, we will get our secret from vault with our DBSettings class:

from superapp.config.database_settings import DBSettings

db_settings = DBSettings()
pprint(db_settings.dict())
# -> 
# {
#     "host": "testhost",
#     "password": "testpassword",
#     "port": 1234,
#     "user": "testuser"
# }
Enter fullscreen mode Exit fullscreen mode

KafkaSettings - connection to brokers

File kafka_settings.py:

from typing import Dict, Any  

from pydantic import Field  

from sitri.settings.contrib.vault import VaultKVSettings  

from superapp.config.provider_config import provider, configurator  


class KafkaSettings(VaultKVSettings):  
    auth_mechanism: str = Field(...)  
    brokers: str = Field(...)  
    auth_data: Dict[str, Any] = Field(...)  

    class Config:  
        provider = provider  
        default_secret_path = "kafka"  
        default_mount_point = f"{configurator.get('app_name')}/common"
Enter fullscreen mode Exit fullscreen mode

In this case, let's imagine that there is one kafka instance for different environments of our service, so the secret is stored along the path superapp/common/kafka:

Note: You can also set secret_path or/and mount_point at the field level so that the provider requests specific values from different secrets (if required). Here is a quote with prioritization of the secret path and mount point from the documentation:

Secret path prioritization:

  1. vault_secret_path (Field arg)
  2. default_secret_path (Config class field)
  3. secret_path (provider initialization optional arg)

Mount point prioritization:

  1. vault_mount_point (Field arg)
  2. default_mount_point (Config class field)
  3. mount_point (provider initialization optional arg)
{
  "auth_data": "{\"password\": \"testpassword\", \"username\": \"testuser\"}",
  "auth_mechanism": "SASL_PLAINTEXT",
  "brokers": "kafka://test"
}
Enter fullscreen mode Exit fullscreen mode

Or

{
    "auth_data":
    {
        "password": "testpassword",
        "username": "testuser"
    },
    "brokers": "kafka://test",
    "auth_mechanism": "SASL_PLAINTEXT"
}
Enter fullscreen mode Exit fullscreen mode

Note: VaultKVSettings can understand both json and Dict itself.

As a result, this class of settings will be able to collect data from the secret like this:

{
    "auth_data":
    {
        "password": "testpassword",
        "username": "testuser"
    },
    "brokers": "kafka://test",
    "auth_mechanism": "SASL_PLAINTEXT"
}
Enter fullscreen mode Exit fullscreen mode

FaustSettings - global configure faust and individual agents

File faust_settings.py:

from typing import Dict  

from pydantic import Field, BaseModel  

from sitri.settings.contrib.vault import VaultKVSettings  

from superapp.config.provider_config import provider  


class AgentConfig(BaseModel):  
    partitions: int = Field(...)  
    concurrency: int = Field(...)  


class FaustSettings(VaultKVSettings):  
    app_name: str = Field(...)  
    default_partitions_count: int = Field(..., vault_secret_key="partitions_count")  
    default_concurrency: int = Field(..., vault_secret_key="agent_concurrency")  
    agents: Dict[str, AgentConfig] = Field(default=None, vault_secret_key="agents_specification")  

    class Config:  
        provider = provider  
        default_secret_path = "faust"
Enter fullscreen mode Exit fullscreen mode

Secret superapp/dev/faust:

{
  "agent_concurrency": "5",
  "app_name": "superapp-workers",
  "partitions_count": "10"
}
Enter fullscreen mode Exit fullscreen mode

If our secret is written as indicated above, then individual agents will take information about the number of partitions in their topics and concurrency from the default fields.

{
  "agents": None,
  "app_name": "superapp-workers",
  "default_concurrency": 5,
  "default_partitions_count": 10
}
Enter fullscreen mode Exit fullscreen mode

However, with the help of the AgentConfig model, we can set individual values for a specific agent. For example, if we have agent X with 5 partitions and concurrency 2, then we can change our secret so that information about this agent is in the "agents" field.

Secret superapp/dev/faust:

{
  "agent_concurrency": "5",
  "agents_specification": {
    "X": {
      "concurrency": "2",
      "partitions": "5"
    }
  },
  "app_name": "superapp-workers",
  "partitions_count": "10"
}
Enter fullscreen mode Exit fullscreen mode

If we initialize our settings now, we will receive, in addition to the default fields, also the configuration for a specific agent:

{
    "agents":
    {
        "X":
        {
            "concurrency": 2,
            "partitions": 5
        }
    },
    "app_name": "superapp-workers",
    "default_concurrency": 5,
    "default_partitions_count": 10
}
Enter fullscreen mode Exit fullscreen mode

Combine the settings classes into a single configuration model

To make our configs even more convenient to use, let's combine all our settings classes into one model by applying the default_factory:

File app_settings.py:

from pydantic import BaseModel, Field  

from superapp.config.database_settings import DBSettings  
from superapp.config.faust_settings import FaustSettings  
from superapp.config.kafka_settings import KafkaSettings  


class AppSettings(BaseModel):  
    db: DBSettings = Field(default_factory=DBSettings)  
    faust: FaustSettings = Field(default_factory=FaustSettings)  
    kafka: KafkaSettings = Field(default_factory=KafkaSettings)
Enter fullscreen mode Exit fullscreen mode

Now, let's go to the main.py file and test the collection of complete configuration data for our application:

from superapp.config import AppSettings  

config = AppSettings()  

print(config)  
print(config.dict())
Enter fullscreen mode Exit fullscreen mode

We get the output of application configuration:

db=DBSettings(user='testuser', password='testpassword', host='testhost', port=1234) 
faust=FaustSettings(app_name='superapp-workers', default_partitions_count=10, default_concurrency=5, agents={'X': AgentConfig(partitions=5, concurrency=2)}) 
kafka=KafkaSettings(auth_mechanism='SASL_PLAINTEXT', brokers='kafka://test', auth_data={'password': 'testpassword', 'username': 'testuser'})
Enter fullscreen mode Exit fullscreen mode
{
    "db":
    {
        "host": "testhost",
        "password": "testpassword",
        "port": 1234,
        "user": "testuser"
    },
    "faust":
    {
        "agents":
        {
            "X":
            {
                "concurrency": 2,
                "partitions": 5
            }
        },
        "app_name": "superapp-workers",
        "default_concurrency": 5,
        "default_partitions_count": 10
    },
    "kafka":
    {
        "auth_data":
        {
            "password": "testpassword",
            "username": "testuser"
        },
        "brokers": "kafka://test",
        "auth_mechanism": "SASL_PLAINTEXT"
    }
}
Enter fullscreen mode Exit fullscreen mode

Happiness, joy, delight!

Afterword

As you can see, the configuring is quite simple with Sitri, after it we get a clear configuration scheme with the required data types for the values, even if they were stored in strings in the vault by default.

Write comments about lib, code or general impressions. I will be glad to any feedback!

P.S. I have uploaded the code from the article to github

Top comments (0)