Merge pull request #167 from uriyyo/feature/piccolo-integration

🔧  Add piccolo integration

Author: uriyyo

README.md

@@ -26,6 +26,7 @@ Available integrations:
 * [orm](https://github.com/encode/orm)
 * [tortoise](https://github.com/tortoise/tortoise-orm)
 * [django](https://github.com/django/django)
+* [piccolo](https://github.com/piccolo-orm/piccolo)
 
 ## Example
 

examples/pagination_piccolo.py

@@ -0,0 +1,87 @@
+import os
+from itertools import count
+from pathlib import Path
+from typing import Any
+
+import uvicorn
+from faker import Faker
+from fastapi import FastAPI
+from piccolo.columns import Integer, Text
+from piccolo.conf.apps import AppConfig, AppRegistry
+from piccolo.engine import SQLiteEngine, engine_finder
+from piccolo.table import Table
+from pydantic import BaseModel
+
+from fastapi_pagination import LimitOffsetPage, Page, add_pagination
+from fastapi_pagination.ext.piccolo import paginate
+
+os.environ["PICCOLO_CONF"] = __name__
+
+_counter = count().__next__
+
+faker = Faker()
+
+
+class _User(Table):
+    id = Integer(default=_counter, primary_key=True)
+    name = Text(required=False, null=True)
+    email = Text(required=False, null=True)
+
+
+class UserIn(BaseModel):
+    name: str
+    email: str
+
+
+class UserOut(UserIn):
+    id: int
+
+    class Config:
+        orm_mode = True
+
+
+app = FastAPI()
+
+DB = SQLiteEngine()
+APP_REGISTRY = AppRegistry()
+
+APP_CONFIG = AppConfig(
+    app_name="example",
+    migrations_folder_path=None,
+    table_classes=[_User],
+)
+
+
+@app.on_event("startup")
+async def on_startup() -> None:
+    engine: SQLiteEngine = engine_finder()
+
+    p = Path(engine.path)
+    if p.exists():
+        os.remove(p)
+
+    await engine.prep_database()
+    await _User.create_table().run()
+
+    for _ in range(100):
+        await _User(
+            name=faker.name(),
+            email=faker.email(),
+        ).save()
+
+
+@app.post("/users", response_model=UserOut)
+async def create_user(user_in: UserIn) -> Any:
+    return await _User(**user_in.dict()).save()
+
+
+@app.get("/users/default", response_model=Page[UserOut])
+@app.get("/users/limit-offset", response_model=LimitOffsetPage[UserOut])
+async def get_users() -> Any:
+    return await paginate(_User)
+
+
+add_pagination(app)
+
+if __name__ == "__main__":
+    uvicorn.run("pagination_piccolo:app")

fastapi_pagination/ext/piccolo.py

@@ -0,0 +1,32 @@
+from copy import deepcopy
+from typing import Optional, Type, Union
+
+from piccolo.query import Select
+from piccolo.query.methods.select import Count
+from piccolo.table import Table
+
+from ..api import create_page, resolve_params
+from ..bases import AbstractPage, AbstractParams
+
+
+async def paginate(query: Union[Select, Type[Table]], params: Optional[AbstractParams] = None) -> AbstractPage:
+    if not isinstance(query, Select):
+        query = query.select()
+
+    # query object is mutable, so we need deepcopy of it
+    query = deepcopy(query)
+
+    params = resolve_params(params)
+    raw_params = params.to_raw_params()
+
+    # need another copy for count query
+    count_query = deepcopy(query)
+    count_query.columns_delegate.selected_columns.clear()
+
+    total = (await count_query.columns(Count()).first())["count"]
+    items = await query.offset(raw_params.offset).limit(raw_params.limit)
+
+    return create_page(items, total, params)
+
+
+__all__ = ["paginate"]