Getting started with Python

Create a new directory, or use an existing project. By default, norm init creates a Python/Postgres target:

norm init

This writes a default norm.yaml at the project root and creates norm_in/ with an empty schema.sql.

Setting up

After norm init, your config matches the default template: async Python, Pydantic models, and max_params: 10.

For this tutorial, set max_params: 1. That makes create_user receive a generated CreateUserParams model instead of separate arguments:

norm.yaml

version: "1"

targets:
  - name: default

    sql:
      engine: postgres
      db_schema: ./norm_in/schema.sql

      repositories: ./norm_in/repositories

    gen:
      out: ./norm_out
      python:
        asynchronous: true
        models: pydantic
        max_params: 1

See max_params for the grouping rules.

If you change the db_schema or repositories paths, run norm init again to create any missing directories and files.

Schema and queries

Start by defining the database schema in schema.sql:

schema.sql

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name text NOT NULL,
    blocked bool DEFAULT false
);

Next, create a repository SQL file under repositories:

users_repo.sql

-- repo_name: UsersRepo

-- name: get_user :one
SELECT * FROM users
WHERE id = :id;

-- name: list_users :many
SELECT * FROM users
ORDER BY name;

-- name: create_user :exec
INSERT INTO users (
  name, blocked
) VALUES (
  :name, :blocked
);

-- name: delete_user :exec
DELETE FROM users
WHERE id = :id;

Generating code

Generate the Python code:

norm generate

Generated output:

├── norm.yaml
└── norm_out/
    ├── __init__.py
    ├── models.py
    └── users_repo.py

models.py

"""
Generated by nORM. READ ONLY.
https://github.com/devfros/norm
"""

from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str
    blocked: bool | None

users_repo.py

"""
Generated by nORM. READ ONLY.
https://github.com/devfros/norm
"""

from psycopg import AsyncConnection as Connection
from pydantic import BaseModel
from .models import User


# Query param schemas
class CreateUserParams(BaseModel):
    name: str
    blocked: bool | None


class UsersRepo:
    def __init__(self, db: Connection) -> None:
        self.db = db

    async def get_user(
        self,
        id: int,
    ) -> User | None:
        query = """
            SELECT
              users.id AS id,
              users.name AS name,
              users.blocked AS blocked
            FROM users
            WHERE
              users.id = %(id)s
        """

        params = {
            "id": id,
        }

        async with self.db.cursor() as cur:
            await cur.execute(query, params)
            result = await cur.fetchone()
            if result is None:
                return None

            result = User(
                id=result[0],
                name=result[1],
                blocked=result[2],
            )

        return result

    async def list_users(
        self,
    ) -> list[User]:
        query = """
            SELECT
              users.id AS id,
              users.name AS name,
              users.blocked AS blocked
            FROM users
            ORDER BY
              users.name
        """

        async with self.db.cursor() as cur:
            await cur.execute(query)
            result = [
                User(
                    id=item[0],
                    name=item[1],
                    blocked=item[2],
                )
                async for item in cur
            ]

        return result

    async def create_user(
        self,
        arg: CreateUserParams,
    ) -> None:
        query = """
            INSERT INTO users (
              name,
              blocked
            )
            VALUES
              (%(name)s, %(blocked)s)
        """

        params = {
            "name": arg.name,
            "blocked": arg.blocked,
        }

        async with self.db.cursor() as cur:
            await cur.execute(query, params)

            result = None

        return result

    async def delete_user(
        self,
        id: int,
    ) -> None:
        query = """
            DELETE FROM users
            WHERE
              id = %(id)s
        """

        params = {
            "id": id,
        }

        async with self.db.cursor() as cur:
            await cur.execute(query, params)

            result = None

        return result

Because create_user has two parameters and this tutorial uses max_params: 1, nORM groups those inputs into CreateUserParams.

Using generated code

usage.py

from norm_out.users_repo import UsersRepo, CreateUserParams

from contextlib import asynccontextmanager

from psycopg import AsyncConnection


@asynccontextmanager
async def get_db():
    connection = await AsyncConnection.connect(conn_str)
    try:
        yield connection
    finally:
        await connection.close()


async def main():
    async with get_db() as db:
        repo = UsersRepo(db=db)

        one = await repo.get_user(id=5)

        many = await repo.list_users()

        await repo.create_user(
            CreateUserParams(
                name="John Doe",
                blocked=False,
            )
        )

        await repo.delete_user(id=5)

Other SQL engines

Change sql.engine in norm.yaml to sqlite, mysql, clickhouse, or duckdb to target another database. Python generation supports all of them; only norm schema pull is Postgres-only today.

See Database and language support and Configuration.