repository

Name: repository
Author: ocbunknown

32stars4forks

How to install repository?

ocbunknown/fastapi-claude-template/repository

$npx -y skills add ocbunknown/fastapi-claude-template --skill repository

Installs into the current project.

Summary

Use when creating or extending a repository under src/database/psql/repositories/. Enforces the project's fixed repository vocabulary (create, select, select_many, update, delete, exists, count, upsert) — no invented methods. Shows exactly how to wrap CRUDRepository, use @on_inte

Files · 1

View on GitHub

SKILL.md

1	# Writing repositories (`src/database/psql/repositories/`)
2
3	Repositories are thin wrappers around `CRUDRepository` that add domain semantics (named arguments, `loads`, ordering, `@on_integrity` for unique constraints). The CRUD verbs are fixed — do not invent new method names.
4
5	## Allowed method vocabulary — memorize this list
6
7	A repository may expose only these verbs (one per responsibility):
8
9	\| Method \| Purpose \| Return \|
10	\|---\|---\|---\|
11	\| `create(**data)` \| insert one row \| `Result[M]` \|
12	\| `select(loads, *filters)` \| fetch one row by identifier(s) \| `Result[M]` \|
13	\| `select_many(loads, *filters, order_by, offset, limit)` \| paginated list \| `Result[tuple[int, Sequence[M]]]` \|
14	\| `update(uuid, /, **data)` \| update one row by id \| `Result[M]` \|
15	\| `delete(**filters)` \| delete one row by id \| `Result[M]` \|
16	\| `exists(**filters)` \| existence check \| `Result[bool]` \|
17	\| `count(**filters)` \| count matching rows \| `Result[int]` \|
18	\| `upsert(conflict_cols, *data)` \| insert-or-update on conflict \| `Result[M]` \|
19
20	Do not invent verbs like `get`, `fetch`, `find_one`, `find_by_email`, `list_all`, `save`, `remove`, `get_or_create`, `paginate`, `search`. If you need "find user by email", that's still `select(login=...)` with a new named parameter. If the current method doesn't support a filter you need, add a new keyword arg to the existing method — do not add a new method.
21
22	The only acceptable additions beyond this vocabulary are domain-specific bulk variants that mirror CRUD (`insert_many` on CRUDRepository already exists — use it via `self._crud.insert_many(...)`).
23
24	## Anatomy of a repository
25
26	Every repository inherits `BaseRepository[models.X]` and accesses CRUD primitives through `self._crud`:
27
28	```python
29	# src/database/psql/repositories/widget.py
30	from collections.abc import Sequence
31	from typing import Optional, Unpack
32
33	import uuid_utils.compat as uuid
34	from sqlalchemy import ColumnExpressionArgument
35
36	import src.database.psql.models as models
37	from src.database.psql.exceptions import InvalidParamsError
38	from src.database.psql.repositories import Result
39	from src.database.psql.repositories.base import BaseRepository
40	from src.database.psql.tools import (
41	on_integrity,
42	sqla_offset_query,
43	sqla_select,
44	unique_scalars,
45	)
46	from src.database.psql.types import OrderBy
47	from src.database.psql.types.widget import (
48	CreateWidgetType,
49	UpdateWidgetType,
50	WidgetLoads,
51	)
52
53
54	class WidgetRepository(BaseRepository[models.Widget]):
55	__slots__ = ()
56
57	@on_integrity("name")
58	async def create(self, **data: Unpack[CreateWidgetType]) -> Result[models.Widget]:
59	return Result("create", await self._crud.insert(**data))
60
61	async def select(
62	self,
63	*loads: WidgetLoads,
64	widget_uuid: Optional[uuid.UUID] = None,
65	name: Optional[str] = None,
66	) -> Result[models.Widget]:
67	if not any([widget_uuid, name]):
68	raise InvalidParamsError("at least one identifier must be provided")
69
70	where_clauses: list[ColumnExpressionArgument[bool]] = []
71	if widget_uuid:
72	where_clauses.append(self.model.uuid == widget_uuid)
73	if name:
74	where_clauses.append(self.model.name == name)
75
76	stmt = sqla_select(model=self.model, loads=loads).where(*where_clauses)
77	return Result(
78	"select", unique_scalars(await self._session.execute(stmt)).first()
79	)
80
81	@on_integrity("name")
82	async def update(
83	self,
84	uuid: uuid.UUID,
85	/,
86	**data: Unpack[UpdateWidgetType],
87	) -> Result[models.Widget]:
88	result = await self._crud.update(self.model.uuid == uuid, **data)
89	return Result("update", result[0] if result else None)
90
91	async def delete(
92	self, widget_uuid: Optional[uuid.UUID] = None
93	) -> Result[models.Widget]:
94	if not widget_uuid:
95	raise InvalidParamsError("at least one identifier must be provided")
96
97	result = await self._crud.delete(self.model.uuid == widget_uuid)
98	return Result("delete", result[0] if result else None)
99
100	async def select_many(
101	self,
102	*loads: WidgetLoads,
103	name: Optional[str] = None,
104	order_by: OrderBy = "desc",
105	offset: int = 0,
106	limit: Optional[int] = None,
107	) -> Result[tuple[int, Sequence[models.Widget]]]:
108	where_clauses: list[ColumnExpressionArgument[bool]] = []
109
110	if name:
111	where_clauses.append(self.model.name.ilike(f"%{name}%"))
112
113	total = await self._crud.count(*where_clauses)
114	if total <= 0:
115	return Result("select", (total, []))

People who liked this also viewed

6 picks

Type

repository

32stars4forks

View on GitHub

How to install repository?

ocbunknown/fastapi-claude-template/repository

$npx -y skills add ocbunknown/fastapi-claude-template --skill repository

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# Writing repositories (`src/database/psql/repositories/`)
2
3	Repositories are thin wrappers around `CRUDRepository` that add domain semantics (named arguments, `loads`, ordering, `@on_integrity` for unique constraints). The CRUD verbs are fixed — do not invent new method names.
4
5	## Allowed method vocabulary — memorize this list
6
7	A repository may expose only these verbs (one per responsibility):
8
9	\| Method \| Purpose \| Return \|
10	\|---\|---\|---\|
11	\| `create(**data)` \| insert one row \| `Result[M]` \|
12	\| `select(loads, *filters)` \| fetch one row by identifier(s) \| `Result[M]` \|
13	\| `select_many(loads, *filters, order_by, offset, limit)` \| paginated list \| `Result[tuple[int, Sequence[M]]]` \|
14	\| `update(uuid, /, **data)` \| update one row by id \| `Result[M]` \|
15	\| `delete(**filters)` \| delete one row by id \| `Result[M]` \|
16	\| `exists(**filters)` \| existence check \| `Result[bool]` \|
17	\| `count(**filters)` \| count matching rows \| `Result[int]` \|
18	\| `upsert(conflict_cols, *data)` \| insert-or-update on conflict \| `Result[M]` \|
19
20	Do not invent verbs like `get`, `fetch`, `find_one`, `find_by_email`, `list_all`, `save`, `remove`, `get_or_create`, `paginate`, `search`. If you need "find user by email", that's still `select(login=...)` with a new named parameter. If the current method doesn't support a filter you need, add a new keyword arg to the existing method — do not add a new method.
21
22	The only acceptable additions beyond this vocabulary are domain-specific bulk variants that mirror CRUD (`insert_many` on CRUDRepository already exists — use it via `self._crud.insert_many(...)`).
23
24	## Anatomy of a repository
25
26	Every repository inherits `BaseRepository[models.X]` and accesses CRUD primitives through `self._crud`:
27
28	```python
29	# src/database/psql/repositories/widget.py
30	from collections.abc import Sequence
31	from typing import Optional, Unpack
32
33	import uuid_utils.compat as uuid
34	from sqlalchemy import ColumnExpressionArgument
35
36	import src.database.psql.models as models
37	from src.database.psql.exceptions import InvalidParamsError
38	from src.database.psql.repositories import Result
39	from src.database.psql.repositories.base import BaseRepository
40	from src.database.psql.tools import (
41	on_integrity,
42	sqla_offset_query,
43	sqla_select,
44	unique_scalars,
45	)
46	from src.database.psql.types import OrderBy
47	from src.database.psql.types.widget import (
48	CreateWidgetType,
49	UpdateWidgetType,
50	WidgetLoads,
51	)
52
53
54	class WidgetRepository(BaseRepository[models.Widget]):
55	__slots__ = ()
56
57	@on_integrity("name")
58	async def create(self, **data: Unpack[CreateWidgetType]) -> Result[models.Widget]:
59	return Result("create", await self._crud.insert(**data))
60
61	async def select(
62	self,
63	*loads: WidgetLoads,
64	widget_uuid: Optional[uuid.UUID] = None,
65	name: Optional[str] = None,
66	) -> Result[models.Widget]:
67	if not any([widget_uuid, name]):
68	raise InvalidParamsError("at least one identifier must be provided")
69
70	where_clauses: list[ColumnExpressionArgument[bool]] = []
71	if widget_uuid:
72	where_clauses.append(self.model.uuid == widget_uuid)
73	if name:
74	where_clauses.append(self.model.name == name)
75
76	stmt = sqla_select(model=self.model, loads=loads).where(*where_clauses)
77	return Result(
78	"select", unique_scalars(await self._session.execute(stmt)).first()
79	)
80
81	@on_integrity("name")
82	async def update(
83	self,
84	uuid: uuid.UUID,
85	/,
86	**data: Unpack[UpdateWidgetType],
87	) -> Result[models.Widget]:
88	result = await self._crud.update(self.model.uuid == uuid, **data)
89	return Result("update", result[0] if result else None)
90
91	async def delete(
92	self, widget_uuid: Optional[uuid.UUID] = None
93	) -> Result[models.Widget]:
94	if not widget_uuid:
95	raise InvalidParamsError("at least one identifier must be provided")
96
97	result = await self._crud.delete(self.model.uuid == widget_uuid)
98	return Result("delete", result[0] if result else None)
99
100	async def select_many(
101	self,
102	*loads: WidgetLoads,
103	name: Optional[str] = None,
104	order_by: OrderBy = "desc",
105	offset: int = 0,
106	limit: Optional[int] = None,
107	) -> Result[tuple[int, Sequence[models.Widget]]]:
108	where_clauses: list[ColumnExpressionArgument[bool]] = []
109
110	if name:
111	where_clauses.append(self.model.name.ilike(f"%{name}%"))
112
113	total = await self._crud.count(*where_clauses)
114	if total <= 0:
115	return Result("select", (total, []))

People who liked this also viewed

6 picks

Type