QOL improvement

Author: Brandon689

GeneratedCode/blog/__pycache__/database.cpython-312.pyc

@@ -0,0 +1,32 @@
+import sqlite3
+from typing import List, Dict, Any
+
+DB_NAME = "C:/2024/i/recoil/blog.db"
+
+def execute_query(query: str, params: tuple = ()) -> List[Dict[str, Any]]:
+    conn = sqlite3.connect(DB_NAME)
+    conn.row_factory = sqlite3.Row
+    cursor = conn.cursor()
+    cursor.execute(query, params)
+    result = [dict(row) for row in cursor.fetchall()]
+    conn.commit()
+    conn.close()
+    return result
+
+def execute_insert(query: str, params: tuple = ()) -> int:
+    conn = sqlite3.connect(DB_NAME)
+    cursor = conn.cursor()
+    cursor.execute(query, params)
+    last_id = cursor.lastrowid
+    conn.commit()
+    conn.close()
+    return last_id
+
+def execute_update_delete(query: str, params: tuple = ()) -> int:
+    conn = sqlite3.connect(DB_NAME)
+    cursor = conn.cursor()
+    cursor.execute(query, params)
+    affected_rows = cursor.rowcount
+    conn.commit()
+    conn.close()
+    return affected_rows

GeneratedCode/blog/__pycache__/main.cpython-312.pyc

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+from database import DB_NAME
+from routers import users
+from routers import categories
+from routers import posts
+from routers import comments
+
+app = FastAPI()
+
+# Include routers
+app.include_router(users.router)
+app.include_router(categories.router)
+app.include_router(posts.router)
+app.include_router(comments.router)
+
+@app.get("/")
+async def root():
+    return {"message": "Welcome to the API"}
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

GeneratedCode/blog/__pycache__/models.cpython-312.pyc

@@ -0,0 +1,57 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+class UsersBase(BaseModel):
+    Username: str
+    Email: str
+    PasswordHash: str
+    CreatedAt: str
+    IsAdmin: bool
+    
+
+class UsersCreate(UsersBase):
+    pass
+
+class Users(UsersBase):
+    Id: int
+
+class CategoriesBase(BaseModel):
+    Name: str
+    Description: Optional[str] = None
+    
+
+class CategoriesCreate(CategoriesBase):
+    pass
+
+class Categories(CategoriesBase):
+    Id: int
+
+class PostsBase(BaseModel):
+    Title: str
+    Content: str
+    CreatedAt: str
+    UpdatedAt: Optional[datetime] = None
+    AuthorId: int
+    CategoryId: int
+    
+
+class PostsCreate(PostsBase):
+    pass
+
+class Posts(PostsBase):
+    Id: int
+
+class CommentsBase(BaseModel):
+    Content: str
+    CreatedAt: str
+    PostId: int
+    UserId: int
+    
+
+class CommentsCreate(CommentsBase):
+    pass
+
+class Comments(CommentsBase):
+    Id: int
+

GeneratedCode/blog/database.py

@@ -0,0 +1,58 @@
+from fastapi import APIRouter, HTTPException, Query, Depends
+from typing import List, Optional
+from models import Categories, CategoriesCreate
+from database import execute_query, execute_insert, execute_update_delete
+
+router = APIRouter(
+    prefix="/categories",
+    tags=["Categories"]
+)
+
+@router.get("/", response_model=List[Categories])
+async def get_categories(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000),
+    sort_by: Optional[str] = None,
+    order: Optional[str] = Query(None, pattern="^(asc|desc)$")
+):
+    query = "SELECT * FROM Categories"
+    if sort_by:
+        query += f" ORDER BY {sort_by}"
+        if order:
+            query += f" {order.upper()}"
+    query += " LIMIT ? OFFSET ?"
+    return execute_query(query, (limit, skip))
+
+@router.get("/{item_id}", response_model=Categories)
+async def get_categories_item(item_id: int):
+    query = "SELECT * FROM Categories WHERE Id = ?"
+    result = execute_query(query, (item_id,))
+    if not result:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return result[0]
+
+@router.post("/", response_model=Categories)
+async def create_categories(item: CategoriesCreate):
+    columns = ", ".join(item.dict().keys())
+    placeholders = ", ".join("?" * len(item.dict()))
+    query = f"INSERT INTO Categories ({columns}) VALUES ({placeholders})"
+    last_id = execute_insert(query, tuple(item.dict().values()))
+    return {**item.dict(), "Id": last_id}
+
+@router.put("/{item_id}", response_model=Categories)
+async def update_categories(item_id: int, item: CategoriesCreate):
+    set_clause = ", ".join(f"{k} = ?" for k in item.dict().keys())
+    query = f"UPDATE Categories SET {set_clause} WHERE Id = ?"
+    values = tuple(item.dict().values()) + (item_id,)
+    affected_rows = execute_update_delete(query, values)
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {**item.dict(), "Id": item_id}
+
+@router.delete("/{item_id}")
+async def delete_categories(item_id: int):
+    query = "DELETE FROM Categories WHERE Id = ?"
+    affected_rows = execute_update_delete(query, (item_id,))
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {"message": "Item deleted successfully"}

GeneratedCode/blog/main.py

@@ -0,0 +1,58 @@
+from fastapi import APIRouter, HTTPException, Query, Depends
+from typing import List, Optional
+from models import Comments, CommentsCreate
+from database import execute_query, execute_insert, execute_update_delete
+
+router = APIRouter(
+    prefix="/comments",
+    tags=["Comments"]
+)
+
+@router.get("/", response_model=List[Comments])
+async def get_comments(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000),
+    sort_by: Optional[str] = None,
+    order: Optional[str] = Query(None, pattern="^(asc|desc)$")
+):
+    query = "SELECT * FROM Comments"
+    if sort_by:
+        query += f" ORDER BY {sort_by}"
+        if order:
+            query += f" {order.upper()}"
+    query += " LIMIT ? OFFSET ?"
+    return execute_query(query, (limit, skip))
+
+@router.get("/{item_id}", response_model=Comments)
+async def get_comments_item(item_id: int):
+    query = "SELECT * FROM Comments WHERE Id = ?"
+    result = execute_query(query, (item_id,))
+    if not result:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return result[0]
+
+@router.post("/", response_model=Comments)
+async def create_comments(item: CommentsCreate):
+    columns = ", ".join(item.dict().keys())
+    placeholders = ", ".join("?" * len(item.dict()))
+    query = f"INSERT INTO Comments ({columns}) VALUES ({placeholders})"
+    last_id = execute_insert(query, tuple(item.dict().values()))
+    return {**item.dict(), "Id": last_id}
+
+@router.put("/{item_id}", response_model=Comments)
+async def update_comments(item_id: int, item: CommentsCreate):
+    set_clause = ", ".join(f"{k} = ?" for k in item.dict().keys())
+    query = f"UPDATE Comments SET {set_clause} WHERE Id = ?"
+    values = tuple(item.dict().values()) + (item_id,)
+    affected_rows = execute_update_delete(query, values)
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {**item.dict(), "Id": item_id}
+
+@router.delete("/{item_id}")
+async def delete_comments(item_id: int):
+    query = "DELETE FROM Comments WHERE Id = ?"
+    affected_rows = execute_update_delete(query, (item_id,))
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {"message": "Item deleted successfully"}

GeneratedCode/blog/models.py

@@ -0,0 +1,58 @@
+from fastapi import APIRouter, HTTPException, Query, Depends
+from typing import List, Optional
+from models import Posts, PostsCreate
+from database import execute_query, execute_insert, execute_update_delete
+
+router = APIRouter(
+    prefix="/posts",
+    tags=["Posts"]
+)
+
+@router.get("/", response_model=List[Posts])
+async def get_posts(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000),
+    sort_by: Optional[str] = None,
+    order: Optional[str] = Query(None, pattern="^(asc|desc)$")
+):
+    query = "SELECT * FROM Posts"
+    if sort_by:
+        query += f" ORDER BY {sort_by}"
+        if order:
+            query += f" {order.upper()}"
+    query += " LIMIT ? OFFSET ?"
+    return execute_query(query, (limit, skip))
+
+@router.get("/{item_id}", response_model=Posts)
+async def get_posts_item(item_id: int):
+    query = "SELECT * FROM Posts WHERE Id = ?"
+    result = execute_query(query, (item_id,))
+    if not result:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return result[0]
+
+@router.post("/", response_model=Posts)
+async def create_posts(item: PostsCreate):
+    columns = ", ".join(item.dict().keys())
+    placeholders = ", ".join("?" * len(item.dict()))
+    query = f"INSERT INTO Posts ({columns}) VALUES ({placeholders})"
+    last_id = execute_insert(query, tuple(item.dict().values()))
+    return {**item.dict(), "Id": last_id}
+
+@router.put("/{item_id}", response_model=Posts)
+async def update_posts(item_id: int, item: PostsCreate):
+    set_clause = ", ".join(f"{k} = ?" for k in item.dict().keys())
+    query = f"UPDATE Posts SET {set_clause} WHERE Id = ?"
+    values = tuple(item.dict().values()) + (item_id,)
+    affected_rows = execute_update_delete(query, values)
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {**item.dict(), "Id": item_id}
+
+@router.delete("/{item_id}")
+async def delete_posts(item_id: int):
+    query = "DELETE FROM Posts WHERE Id = ?"
+    affected_rows = execute_update_delete(query, (item_id,))
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {"message": "Item deleted successfully"}

GeneratedCode/blog/routers/__init__.py

@@ -0,0 +1,58 @@
+from fastapi import APIRouter, HTTPException, Query, Depends
+from typing import List, Optional
+from models import Users, UsersCreate
+from database import execute_query, execute_insert, execute_update_delete
+
+router = APIRouter(
+    prefix="/users",
+    tags=["Users"]
+)
+
+@router.get("/", response_model=List[Users])
+async def get_users(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000),
+    sort_by: Optional[str] = None,
+    order: Optional[str] = Query(None, pattern="^(asc|desc)$")
+):
+    query = "SELECT * FROM Users"
+    if sort_by:
+        query += f" ORDER BY {sort_by}"
+        if order:
+            query += f" {order.upper()}"
+    query += " LIMIT ? OFFSET ?"
+    return execute_query(query, (limit, skip))
+
+@router.get("/{item_id}", response_model=Users)
+async def get_users_item(item_id: int):
+    query = "SELECT * FROM Users WHERE Id = ?"
+    result = execute_query(query, (item_id,))
+    if not result:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return result[0]
+
+@router.post("/", response_model=Users)
+async def create_users(item: UsersCreate):
+    columns = ", ".join(item.dict().keys())
+    placeholders = ", ".join("?" * len(item.dict()))
+    query = f"INSERT INTO Users ({columns}) VALUES ({placeholders})"
+    last_id = execute_insert(query, tuple(item.dict().values()))
+    return {**item.dict(), "Id": last_id}
+
+@router.put("/{item_id}", response_model=Users)
+async def update_users(item_id: int, item: UsersCreate):
+    set_clause = ", ".join(f"{k} = ?" for k in item.dict().keys())
+    query = f"UPDATE Users SET {set_clause} WHERE Id = ?"
+    values = tuple(item.dict().values()) + (item_id,)
+    affected_rows = execute_update_delete(query, values)
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {**item.dict(), "Id": item_id}
+
+@router.delete("/{item_id}")
+async def delete_users(item_id: int):
+    query = "DELETE FROM Users WHERE Id = ?"
+    affected_rows = execute_update_delete(query, (item_id,))
+    if affected_rows == 0:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return {"message": "Item deleted successfully"}

GeneratedCode/blog/routers/__pycache__/__init__.cpython-312.pyc

@@ -0,0 +1,32 @@
+import sqlite3
+from typing import List, Dict, Any
+
+DB_NAME = "C:/2024/h/Moshi/Moshi.MyMusic/spotify.db"
+
+def execute_query(query: str, params: tuple = ()) -> List[Dict[str, Any]]:
+    conn = sqlite3.connect(DB_NAME)
+    conn.row_factory = sqlite3.Row
+    cursor = conn.cursor()
+    cursor.execute(query, params)
+    result = [dict(row) for row in cursor.fetchall()]
+    conn.commit()
+    conn.close()
+    return result
+
+def execute_insert(query: str, params: tuple = ()) -> int:
+    conn = sqlite3.connect(DB_NAME)
+    cursor = conn.cursor()
+    cursor.execute(query, params)
+    last_id = cursor.lastrowid
+    conn.commit()
+    conn.close()
+    return last_id
+
+def execute_update_delete(query: str, params: tuple = ()) -> int:
+    conn = sqlite3.connect(DB_NAME)
+    cursor = conn.cursor()
+    cursor.execute(query, params)
+    affected_rows = cursor.rowcount
+    conn.commit()
+    conn.close()
+    return affected_rows