swadhin / justapi public

swadhinbiswas

[email protected]
2026-07-09 16:00:00 +0600 committed NaN years ago
b48e461
commit b48e461
16 files changed
+316 -0
Showing 16 changed files+316-0
examples/01_hello_world.pyexamples/02_path_query_params.pyexamples/03_request_body.pyexamples/04_dependencies.pyexamples/05_middleware.pyexamples/06_error_handling.pyexamples/07_file_uploads.pyexamples/08_routers.pyexamples/09_db_integration.pyexamples/10_websockets.pyexamples/plugin_cache.pytest_project/.gitignoretest_project/README.mdtest_project/app/__init__.pytest_project/app/main.pytest_project/requirements.txt
▼examples/01_hello_world.py
+14-0
@@ -0,0 +1,14 @@
1
+from justapi import JustAPIApp
2
+
3
+app = JustAPIApp()
4
+
5
+@app.get("/")
6
+def read_root():
7
+ return {"Hello": "World"}
8
+
9
+@app.get("/items/{item_id}")
10
+def read_item(item_id: int):
11
+ return {"item_id": item_id, "name": "Item Name"}
12
+
13
+if __name__ == "__main__":
14
+ app.run("127.0.0.1:8000")
0
15
▼examples/02_path_query_params.py
+20-0
@@ -0,0 +1,20 @@
1
+from justapi import JustAPIApp
2
+from typing import Optional
3
+
4
+app = JustAPIApp()
5
+
6
+@app.get("/users/{user_id}")
7
+def get_user(user_id: int, q: Optional[str] = None, skip: int = 0, limit: int = 10):
8
+ """
9
+ Fetch a user by ID, with optional query parameters.
10
+ Try accessing: /users/42?q=search&limit=5
11
+ """
12
+ return {
13
+ "user_id": user_id,
14
+ "q": q,
15
+ "skip": skip,
16
+ "limit": limit
17
+ }
18
+
19
+if __name__ == "__main__":
20
+ app.run("127.0.0.1:8000")
0
21
▼examples/03_request_body.py
+25-0
@@ -0,0 +1,25 @@
1
+from justapi import JustAPIApp
2
+from pydantic import BaseModel
3
+
4
+app = JustAPIApp()
5
+
6
+class Item(BaseModel):
7
+ name: str
8
+ description: str | None = None
9
+ price: float
10
+ tax: float | None = None
11
+
12
+@app.post("/items/")
13
+async def create_item(item: Item):
14
+ """
15
+ Expects a JSON body matching the Item schema.
16
+ Returns the created item with calculated total price.
17
+ """
18
+ item_dict = item.model_dump()
19
+ if item.tax:
20
+ price_with_tax = item.price + item.tax
21
+ item_dict.update({"price_with_tax": price_with_tax})
22
+ return item_dict
23
+
24
+if __name__ == "__main__":
25
+ app.run("127.0.0.1:8000")
0
26
▼examples/04_dependencies.py
+17-0
@@ -0,0 +1,17 @@
1
+from justapi import JustAPIApp, Depends
2
+
3
+app = JustAPIApp()
4
+
5
+def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
6
+ return {"q": q, "skip": skip, "limit": limit}
7
+
8
+@app.get("/items/")
9
+def read_items(commons: dict = Depends(common_parameters)):
10
+ return {"message": "Reading items", "params": commons}
11
+
12
+@app.get("/users/")
13
+def read_users(commons: dict = Depends(common_parameters)):
14
+ return {"message": "Reading users", "params": commons}
15
+
16
+if __name__ == "__main__":
17
+ app.run("127.0.0.1:8000")
0
18
▼examples/05_middleware.py
+28-0
@@ -0,0 +1,28 @@
1
+import time
2
+from justapi import JustAPIApp
3
+
4
+app = JustAPIApp()
5
+
6
+@app.middleware("http")
7
+async def add_process_time_header(request, call_next):
8
+ start_time = time.time()
9
+
10
+ # Process the request down the chain
11
+ response = await call_next(request)
12
+
13
+ # Add a custom header
14
+ process_time = time.time() - start_time
15
+ if "headers" not in response:
16
+ response["headers"] = []
17
+ response["headers"].append((b"X-Process-Time", str(process_time).encode("utf-8")))
18
+
19
+ return response
20
+
21
+@app.get("/")
22
+async def root():
23
+ # Simulate some work
24
+ time.sleep(0.1)
25
+ return {"message": "Hello World with Middleware"}
26
+
27
+if __name__ == "__main__":
28
+ app.run("127.0.0.1:8000")
0
29
▼examples/06_error_handling.py
+26-0
@@ -0,0 +1,26 @@
1
+from justapi import JustAPIApp, HTTPException, RequestValidationError
2
+
3
+app = JustAPIApp()
4
+
5
+@app.exception_handler(HTTPException)
6
+def http_exception_handler(request, exc: HTTPException):
7
+ return {
8
+ "status": exc.status_code,
9
+ "body": {"error": "Custom Error Format", "detail": exc.detail},
10
+ }
11
+
12
+@app.exception_handler(RequestValidationError)
13
+def validation_exception_handler(request, exc: RequestValidationError):
14
+ return {
15
+ "status": 422,
16
+ "body": {"error": "Validation Failed", "issues": exc.errors()},
17
+ }
18
+
19
+@app.get("/items/{item_id}")
20
+def read_item(item_id: int):
21
+ if item_id == 3:
22
+ raise HTTPException(status_code=418, detail="I'm a teapot")
23
+ return {"item_id": item_id}
24
+
25
+if __name__ == "__main__":
26
+ app.run("127.0.0.1:8000")
0
27
▼examples/07_file_uploads.py
+26-0
@@ -0,0 +1,26 @@
1
+import os
2
+from justapi import JustAPIApp, UploadFile, File
3
+
4
+app = JustAPIApp()
5
+
6
+@app.post("/upload/")
7
+async def upload_file(file: UploadFile = File(...)):
8
+ """
9
+ Demonstrates receiving a streamed multipart file.
10
+ The file is saved to a temporary location to avoid OOM issues.
11
+ """
12
+ content_type = file.content_type
13
+ filename = file.filename
14
+ temp_path = file.file
15
+
16
+ file_size = os.path.getsize(temp_path)
17
+
18
+ return {
19
+ "filename": filename,
20
+ "content_type": content_type,
21
+ "size_bytes": file_size,
22
+ "message": f"Successfully uploaded {filename}"
23
+ }
24
+
25
+if __name__ == "__main__":
26
+ app.run("127.0.0.1:8000")
0
27
▼examples/08_routers.py
+18-0
@@ -0,0 +1,18 @@
1
+from justapi import JustAPIApp, APIRouter
2
+
3
+app = JustAPIApp()
4
+router = APIRouter(prefix="/api/v1")
5
+
6
+@router.get("/users/")
7
+def get_users():
8
+ return [{"username": "alice"}, {"username": "bob"}]
9
+
10
+@router.get("/users/{username}")
11
+def get_user(username: str):
12
+ return {"username": username}
13
+
14
+# Include the router in the main application
15
+app.include_router(router)
16
+
17
+if __name__ == "__main__":
18
+ app.run("127.0.0.1:8000")
0
19
▼examples/09_db_integration.py
+31-0
@@ -0,0 +1,31 @@
1
+import asyncio
2
+from justapi import JustAPIApp
3
+from pydantic import BaseModel
4
+
5
+app = JustAPIApp()
6
+
7
+# Mock database
8
+fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
9
+
10
+class Item(BaseModel):
11
+ item_name: str
12
+
13
+@app.post("/items/")
14
+async def create_item(item: Item):
15
+ """
16
+ Simulates inserting a record into a database asynchronously.
17
+ """
18
+ await asyncio.sleep(0.5) # Simulate DB latency
19
+ fake_items_db.append(item.model_dump())
20
+ return item
21
+
22
+@app.get("/items/")
23
+async def read_items(skip: int = 0, limit: int = 10):
24
+ """
25
+ Simulates querying a database.
26
+ """
27
+ await asyncio.sleep(0.1) # Simulate DB latency
28
+ return fake_items_db[skip : skip + limit]
29
+
30
+if __name__ == "__main__":
31
+ app.run("127.0.0.1:8000")
0
32
▼examples/10_websockets.py
+20-0
@@ -0,0 +1,20 @@
1
+import asyncio
2
+from justapi import JustAPIApp
3
+
4
+app = JustAPIApp()
5
+
6
+@app.websocket("/ws")
7
+async def websocket_endpoint(ws):
8
+ """
9
+ A simple WebSocket echo server.
10
+ """
11
+ await ws.accept()
12
+ try:
13
+ while True:
14
+ data = await ws.receive_text()
15
+ await ws.send_text(f"Message text was: {data}")
16
+ except Exception as e:
17
+ print("Client disconnected:", e)
18
+
19
+if __name__ == "__main__":
20
+ app.run("127.0.0.1:8000")
0
21
▼examples/plugin_cache.py
+41-0
@@ -0,0 +1,41 @@
1
+from justapi import JustAPIApp
2
+
3
+class RedisCachePlugin:
4
+ """A simple plugin to cache responses in an in-memory dict (simulating Redis)."""
5
+ def __init__(self):
6
+ self.cache = {}
7
+
8
+ def build(self, app: JustAPIApp):
9
+ print("[RedisCachePlugin] Building plugin...")
10
+ # In a real app we might inject this cache into the app state
11
+
12
+ async def on_startup(self):
13
+ print("[RedisCachePlugin] Connecting to simulated Redis...")
14
+ self.cache.clear()
15
+
16
+ async def on_shutdown(self):
17
+ print("[RedisCachePlugin] Disconnecting simulated Redis...")
18
+ self.cache.clear()
19
+
20
+ def get(self, key):
21
+ return self.cache.get(key)
22
+
23
+ def set(self, key, value):
24
+ self.cache[key] = value
25
+
26
+if __name__ == "__main__":
27
+ app = JustAPIApp()
28
+ cache_plugin = RedisCachePlugin()
29
+ app.use(cache_plugin)
30
+
31
+ def cached_hello(request):
32
+ key = "hello"
33
+ if cached := cache_plugin.get(key):
34
+ return {"message": cached, "cached": True}
35
+
36
+ value = "world"
37
+ cache_plugin.set(key, value)
38
+ return {"message": value, "cached": False}
39
+
40
+ app.get("/hello", cached_hello)
41
+ print("Run this file using a python runner to see it in action.")
0
42
▼test_project/.gitignore
+6-0
@@ -0,0 +1,6 @@
1
+*.pyc
2
+__pycache__/
3
+.env
4
+*.egg-info/
5
+dist/
6
+build/
0
7
▼test_project/README.md
+22-0
@@ -0,0 +1,22 @@
1
+# test_project
2
+
3
+A JustAPI application.
4
+
5
+## Quick start
6
+
7
+```bash
8
+justapi serve
9
+```
10
+
11
+## Development
12
+
13
+```bash
14
+# Install dependencies
15
+pip install -r requirements.txt
16
+
17
+# Run migrations
18
+justapi db migrate --url postgres://localhost/test_project
19
+
20
+# Start server with hot reload
21
+justapi serve --reload
22
+```
0
23
▼test_project/app/__init__.py
+0-0
0
0
▼test_project/app/main.py
+20-0
@@ -0,0 +1,20 @@
1
+"""JustAPI application — test_project
2
+"""
3
+
4
+from justapi import JustAPIApp
5
+
6
+app = JustAPIApp()
7
+
8
+
9
+@app.get("/")
10
+async def root():
11
+ return {"message": "Hello from test_project!"}
12
+
13
+
14
+@app.get("/health")
15
+async def health():
16
+ return {"status": "healthy"}
17
+
18
+
19
+if __name__ == "__main__":
20
+ app.run()
0
21
▼test_project/requirements.txt
+2-0
@@ -0,0 +1,2 @@
1
+# Add your Python dependencies here
2
+# justapi is pre-installed with the runtime
0
3