1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
from __future__ import annotations
import datetime # noqa: TC003
from uuid import UUID # noqa: TC003
from sanic import Sanic
from sqlalchemy import ForeignKey
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import Mapped, mapped_column, relationship
from advanced_alchemy.extensions.sanic import (
AdvancedAlchemy,
AsyncSessionConfig,
SQLAlchemyAsyncConfig,
base,
filters,
repository,
service,
)
# the SQLAlchemy base includes a declarative model for you to use in your models.
# The `Base` class includes a `UUID` based primary key (`id`)
class AuthorModel(base.UUIDBase):
# we can optionally provide the table name instead of auto-generating it
__tablename__ = "author"
name: Mapped[str]
dob: Mapped[datetime.date | None]
books: Mapped[list[BookModel]] = relationship(back_populates="author", lazy="noload")
# The `AuditBase` class includes the same UUID` based primary key (`id`) and 2
# additional columns: `created` and `updated`. `created` is a timestamp of when the
# record created, and `updated` is the last time the record was modified.
class BookModel(base.UUIDAuditBase):
__tablename__ = "book"
title: Mapped[str]
author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
author: Mapped[AuthorModel] = relationship(lazy="joined", innerjoin=True, viewonly=True)
class AuthorService(service.SQLAlchemyAsyncRepositoryService[AuthorModel]):
"""Author service."""
class Repo(repository.SQLAlchemyAsyncRepository[AuthorModel]):
"""Author repository."""
model_type = AuthorModel
repository_type = Repo
# #######################
# Dependencies
# #######################
async def provide_authors_service(db_session: AsyncSession) -> AuthorService:
"""This provides the default Authors repository."""
return AuthorService(session=db_session)
# we can optionally override the default `select` used for the repository to pass in
# specific SQL options such as join details
async def provide_author_details_service(
db_session: AsyncSession,
) -> AuthorService:
"""This provides a simple example demonstrating how to override the join options for the repository."""
return AuthorService(load=[AuthorModel.books], session=db_session)
def provide_limit_offset_pagination(
current_page: int = 1,
page_size: int = 10,
) -> filters.LimitOffset:
"""Add offset/limit pagination.
Return type consumed by `Repository.apply_limit_offset_pagination()`.
Parameters
----------
current_page : int
LIMIT to apply to select.
page_size : int
OFFSET to apply to select.
"""
return filters.LimitOffset(page_size, page_size * (current_page - 1))
# #######################
# Application
# #######################
session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
connection_string="sqlite+aiosqlite:///test.sqlite",
session_config=session_config,
) # Create 'db_session' dependency.
app = Sanic("AlchemySanicApp")
alchemy = AdvancedAlchemy(sqlalchemy_config=sqlalchemy_config)
alchemy.register(app)
alchemy.add_session_dependency(AsyncSession)
|
