Adding Additional Features to the Repository

While most of the functionality you need is built into the repository, there are still cases where you need to add in additional functionality. Let’s explore ways that we can add functionality on top of the repository pattern.

Tip

The full code for this tutorial can be found below in the Full Code section.

Slug Fields

app.py
 1from advanced_alchemy.extensions.litestar import (
 2    base,
 3)
 4from sqlalchemy.orm import Mapped, declarative_mixin, mapped_column
 5from sqlalchemy.types import String
 6
 7
 8
 9@declarative_mixin
10class SlugKey:
11    """Slug unique Field Model Mixin."""
12
13    __abstract__ = True
14    slug: Mapped[str] = mapped_column(String(length=100), nullable=False, unique=True, sort_order=-9)
15
16
17class BlogPost(base.UUIDAuditBase, SlugKey):
18    title: Mapped[str]
19    content: Mapped[str]

In this example, we are using a BlogPost model to hold blog post titles and contents. The primary key for this model is a UUID type. UUID and int are good options for primary keys, but there are a number of reasons you may not want to use them in your routes. For instance, it can be a security problem to expose integer-based primary keys in the URL. While UUIDs don’t have this same problem, they are not user-friendly or easy-to-remember, and create complex URLs. One way to solve this is to add a user friendly unique identifier to the table that can be used for urls. This is often called a “slug”.

First, we’ll create a SlugKey field mixin that adds a text-based, URL-friendly, unique column slug to the table. We want to ensure we create a slug value based on the data passed to the title field. To demonstrate what we are trying to accomplish, we want a record that has a blog title of “Follow the Yellow Brick Road!” to have the slugified value of “follow-the-yellow-brick-road”.

app.py
 1from __future__ import annotations
 2
 3import random
 4import re
 5import string
 6import unicodedata
 7from typing import TYPE_CHECKING, Any
 8from uuid import UUID
 9    base,
10    repository,
11)
12from pydantic import BaseModel as _BaseModel
13from pydantic import TypeAdapter
14from sqlalchemy.orm import Mapped, declarative_mixin, mapped_column
15from sqlalchemy.types import String
16
17from litestar import Litestar, get, post
18from litestar.di import NamedDependency, Provide
19
20
21class SQLAlchemyAsyncSlugRepository(repository.SQLAlchemyAsyncRepository[repository.ModelT]):
22    """Extends the repository to include slug model features.."""
23
24    async def get_available_slug(
25        self,
26        value_to_slugify: str,
27        **kwargs: Any,
28    ) -> str:
29        """Get a unique slug for the supplied value.
30
31        If the value is found to exist, a random 4 digit character is appended to the end.
32        There may be a better way to do this, but I wanted to limit the number of
33        additional database calls.
34
35        Args:
36            value_to_slugify (str): A string that should be converted to a unique slug.
37            **kwargs: stuff
38
39        Returns:
40            str: a unique slug for the supplied value. This is safe for URLs and other
41            unique identifiers.
42        """
43        slug = self._slugify(value_to_slugify)
44        if await self._is_slug_unique(slug):
45            return slug
46        # generate a random 4 digit alphanumeric string to make the slug unique and
47        # avoid another DB lookup.
48        random_string = "".join(random.choices(string.ascii_lowercase + string.digits, k=4))
49        return f"{slug}-{random_string}"
50
51    @staticmethod
52    def _slugify(value: str) -> str:
53        """slugify.
54
55        Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated
56        dashes to single dashes. Remove characters that aren't alphanumerics,
57        underscores, or hyphens. Convert to lowercase. Also strip leading and
58        trailing whitespace, dashes, and underscores.
59
60        Args:
61            value (str): the string to slugify
62
63        Returns:
64            str: a slugified string of the value parameter
65        """
66        value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
67        value = re.sub(r"[^\w\s-]", "", value.lower())
68        return re.sub(r"[-\s]+", "-", value).strip("-_")
69
70    async def _is_slug_unique(
71        self,
72        slug: str,
73        **kwargs: Any,
74    ) -> bool:
75        return await self.get_one_or_none(slug=slug) is None

Since the BlogPost.title field is not marked as unique, this means that we’ll have to test the slug value for uniqueness before the insert. If the initial slug is found, a random set of digits are appended to the end of the slug to make it unique.

app.py
 1from __future__ import annotations
 2
 3from uuid import UUID
 4
 5from advanced_alchemy.extensions.litestar import (
 6    base,
 7)
 8from pydantic import BaseModel as _BaseModel
 9from sqlalchemy.orm import Mapped
10
11from litestar import post
12from litestar.di import NamedDependency
13
14
15class BaseModel(_BaseModel):
16    """Extend Pydantic's BaseModel to enable ORM mode"""
17
18    model_config = {"from_attributes": True}
19
20
21class BlogPost(base.UUIDAuditBase, SlugKey):
22    title: Mapped[str]
23    content: Mapped[str]
24
25
26class BlogPostRepository(SQLAlchemyAsyncSlugRepository[BlogPost]):
27    """Blog Post repository."""
28
29    model_type = BlogPost
30
31
32class BlogPostDTO(BaseModel):
33    id: UUID | None
34    slug: str
35    title: str
36    content: str
37
38
39class BlogPostCreate(BaseModel):
40    title: str
41    content: str
42@post(path="/")
43async def create_blog(
44    blog_post_repo: NamedDependency[BlogPostRepository],
45    data: BlogPostCreate,
46) -> BlogPostDTO:
47    """Create a new blog post."""
48    _data = data.model_dump(exclude_unset=True, by_alias=False, exclude_none=True)
49    _data["slug"] = await blog_post_repo.get_available_slug(_data["title"])
50    obj = await blog_post_repo.add(BlogPost(**_data))
51    await blog_post_repo.session.commit()
52    return BlogPostDTO.model_validate(obj)

We are all set to use this in our routes now. First, we’ll convert our incoming Pydantic model to a dictionary. Next, we’ll fetch a unique slug for our text. Finally, we insert the model with the added slug.

Note

Using this method does introduce an additional query on each insert. This should be considered when determining which fields actually need this type of functionality.

Full Code

Full Code (click to toggle)
app.py
  1from __future__ import annotations
  2
  3import random
  4import re
  5import string
  6import unicodedata
  7from typing import TYPE_CHECKING, Any
  8from uuid import UUID
  9
 10from advanced_alchemy.extensions.litestar import (
 11    AsyncSessionConfig,
 12    SQLAlchemyAsyncConfig,
 13    SQLAlchemyInitPlugin,
 14    base,
 15    repository,
 16)
 17from pydantic import BaseModel as _BaseModel
 18from pydantic import TypeAdapter
 19from sqlalchemy.orm import Mapped, declarative_mixin, mapped_column
 20from sqlalchemy.types import String
 21
 22from litestar import Litestar, get, post
 23from litestar.di import NamedDependency, Provide
 24from litestar.params import FromPath
 25
 26if TYPE_CHECKING:
 27    from sqlalchemy.ext.asyncio import AsyncSession
 28
 29
 30class BaseModel(_BaseModel):
 31    """Extend Pydantic's BaseModel to enable ORM mode"""
 32
 33    model_config = {"from_attributes": True}
 34
 35
 36# we are going to add a simple "slug" to our model that is a URL safe surrogate key to
 37# our database record.
 38@declarative_mixin
 39class SlugKey:
 40    """Slug unique Field Model Mixin."""
 41
 42    __abstract__ = True
 43    slug: Mapped[str] = mapped_column(String(length=100), nullable=False, unique=True, sort_order=-9)
 44
 45
 46# this class can be re-used with any model that has the `SlugKey` Mixin
 47class SQLAlchemyAsyncSlugRepository(repository.SQLAlchemyAsyncRepository[repository.ModelT]):
 48    """Extends the repository to include slug model features.."""
 49
 50    async def get_available_slug(
 51        self,
 52        value_to_slugify: str,
 53        **kwargs: Any,
 54    ) -> str:
 55        """Get a unique slug for the supplied value.
 56
 57        If the value is found to exist, a random 4 digit character is appended to the end.
 58        There may be a better way to do this, but I wanted to limit the number of
 59        additional database calls.
 60
 61        Args:
 62            value_to_slugify (str): A string that should be converted to a unique slug.
 63            **kwargs: stuff
 64
 65        Returns:
 66            str: a unique slug for the supplied value. This is safe for URLs and other
 67            unique identifiers.
 68        """
 69        slug = self._slugify(value_to_slugify)
 70        if await self._is_slug_unique(slug):
 71            return slug
 72        # generate a random 4 digit alphanumeric string to make the slug unique and
 73        # avoid another DB lookup.
 74        random_string = "".join(random.choices(string.ascii_lowercase + string.digits, k=4))
 75        return f"{slug}-{random_string}"
 76
 77    @staticmethod
 78    def _slugify(value: str) -> str:
 79        """slugify.
 80
 81        Convert to ASCII if 'allow_unicode' is False. Convert spaces or repeated
 82        dashes to single dashes. Remove characters that aren't alphanumerics,
 83        underscores, or hyphens. Convert to lowercase. Also strip leading and
 84        trailing whitespace, dashes, and underscores.
 85
 86        Args:
 87            value (str): the string to slugify
 88
 89        Returns:
 90            str: a slugified string of the value parameter
 91        """
 92        value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
 93        value = re.sub(r"[^\w\s-]", "", value.lower())
 94        return re.sub(r"[-\s]+", "-", value).strip("-_")
 95
 96    async def _is_slug_unique(
 97        self,
 98        slug: str,
 99        **kwargs: Any,
100    ) -> bool:
101        return await self.get_one_or_none(slug=slug) is None
102
103
104# The `UUIDAuditBase` class includes the same UUID` based primary key (`id`) and 2
105# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
106# record created, and `updated_at` is the last time the record was modified.
107class BlogPost(base.UUIDAuditBase, SlugKey):
108    title: Mapped[str]
109    content: Mapped[str]
110
111
112class BlogPostRepository(SQLAlchemyAsyncSlugRepository[BlogPost]):
113    """Blog Post repository."""
114
115    model_type = BlogPost
116
117
118class BlogPostDTO(BaseModel):
119    id: UUID | None
120    slug: str
121    title: str
122    content: str
123
124
125class BlogPostCreate(BaseModel):
126    title: str
127    content: str
128
129
130# we can optionally override the default `select` used for the repository to pass in
131# specific SQL options such as join details
132async def provide_blog_post_repo(db_session: NamedDependency[AsyncSession]) -> BlogPostRepository:
133    """This provides a simple example demonstrating how to override the join options
134    for the repository."""
135    return BlogPostRepository(session=db_session)
136
137
138session_config = AsyncSessionConfig(expire_on_commit=False)
139sqlalchemy_config = SQLAlchemyAsyncConfig(
140    connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config
141)  # Create 'async_session' dependency.
142sqlalchemy_plugin = SQLAlchemyInitPlugin(config=sqlalchemy_config)
143
144
145async def on_startup() -> None:
146    """Initializes the database."""
147    async with sqlalchemy_config.get_engine().begin() as conn:
148        await conn.run_sync(base.UUIDAuditBase.metadata.create_all)
149
150
151@get(path="/")
152async def get_blogs(
153    blog_post_repo: NamedDependency[BlogPostRepository],
154) -> list[BlogPostDTO]:
155    """Interact with SQLAlchemy engine and session."""
156    objs = await blog_post_repo.list()
157    type_adapter = TypeAdapter(list[BlogPostDTO])
158    return type_adapter.validate_python(objs)
159
160
161@get(path="/{post_slug:str}")
162async def get_blog_details(
163    post_slug: FromPath[str],
164    blog_post_repo: NamedDependency[BlogPostRepository],
165) -> BlogPostDTO:
166    """Interact with SQLAlchemy engine and session."""
167    obj = await blog_post_repo.get_one(slug=post_slug)
168    return BlogPostDTO.model_validate(obj)
169
170
171@post(path="/")
172async def create_blog(
173    blog_post_repo: NamedDependency[BlogPostRepository],
174    data: BlogPostCreate,
175) -> BlogPostDTO:
176    """Create a new blog post."""
177    _data = data.model_dump(exclude_unset=True, by_alias=False, exclude_none=True)
178    _data["slug"] = await blog_post_repo.get_available_slug(_data["title"])
179    obj = await blog_post_repo.add(BlogPost(**_data))
180    await blog_post_repo.session.commit()
181    return BlogPostDTO.model_validate(obj)
182
183
184app = Litestar(
185    route_handlers=[create_blog, get_blogs, get_blog_details],
186    dependencies={"blog_post_repo": Provide(provide_blog_post_repo, sync_to_thread=False)},
187    on_startup=[on_startup],
188    plugins=[SQLAlchemyInitPlugin(config=sqlalchemy_config)],
189)