backend.im/taskmanagerapi-ttkjqk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add user authentication system with JWT tokens

Browse Source 
- Add user model with relationship to tasks
- Implement JWT token authentication
- Create user registration and login endpoints
- Update task endpoints to filter by current user
- Add Alembic migration for user table
- Update documentation with authentication details
This commit is contained in:
Automated Action 2025-05-16 12:40:03 +00:00
parent a4cc5dbcca
commit 0ceeef31a6
26 changed files with 1237 additions and 476 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
README.md
View File

@ -4,7 +4,9 @@ A RESTful API for managing tasks, built with FastAPI and SQLite.
## Features
- Task CRUD operations
- User authentication with JWT tokens
- User registration and login
- Task CRUD operations with user-based access control
- Task status and priority management
- Task completion tracking
- API documentation with Swagger UI and ReDoc
@ -18,6 +20,9 @@ A RESTful API for managing tasks, built with FastAPI and SQLite.
- SQLite: Lightweight relational database
- Pydantic: Data validation and settings management
- Uvicorn: ASGI server for FastAPI applications
- JWT: JSON Web Tokens for authentication
- Passlib: Password hashing and verification
- Python-Jose: Python implementation of JWT
## API Endpoints
@ -25,14 +30,21 @@ A RESTful API for managing tasks, built with FastAPI and SQLite.
- `GET /`: Get API information and available endpoints
### Task Management
### Authentication
- `GET /tasks`: Get all tasks
- `POST /tasks`: Create a new task
- `GET /tasks/{task_id}`: Get a specific task
- `PUT /tasks/{task_id}`: Update a task
- `DELETE /tasks/{task_id}`: Delete a task
- `POST /tasks/{task_id}/complete`: Mark a task as completed
- `POST /auth/register`: Register a new user
- `POST /auth/login`: Login to get access token
- `GET /auth/me`: Get current user information
- `POST /auth/test-token`: Test if the access token is valid
### Task Management (requires authentication)
- `GET /tasks`: Get all tasks for the current user
- `POST /tasks`: Create a new task for the current user
- `GET /tasks/{task_id}`: Get a specific task for the current user
- `PUT /tasks/{task_id}`: Update a task for the current user
- `DELETE /tasks/{task_id}`: Delete a task for the current user
- `POST /tasks/{task_id}/complete`: Mark a task as completed for the current user
### Health and Diagnostic Endpoints
@ -41,12 +53,37 @@ A RESTful API for managing tasks, built with FastAPI and SQLite.
## Example Curl Commands
### Create a new task
### Register a new user
```bash
curl -X 'POST' \
'https://taskmanagerapi-ttkjqk.backend.im/auth/register' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"email": "user@example.com",
"username": "testuser",
"password": "password123"
}'
```
### Login to get access token
```bash
curl -X 'POST' \
'https://taskmanagerapi-ttkjqk.backend.im/auth/login' \
-H 'accept: application/json' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-d 'username=user@example.com&password=password123'
```
### Create a new task (with authentication)
```bash
curl -X 'POST' \
'https://taskmanagerapi-ttkjqk.backend.im/tasks/' \
-H 'accept: application/json' \
-H 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
-H 'Content-Type: application/json' \
-d '{
"title": "Complete project documentation",
@ -58,28 +95,31 @@ curl -X 'POST' \
}'
```
### Get all tasks
### Get all tasks (with authentication)
```bash
curl -X 'GET' \
'https://taskmanagerapi-ttkjqk.backend.im/tasks/' \
-H 'accept: application/json'
-H 'accept: application/json' \
-H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
```
### Get a specific task
### Get a specific task (with authentication)
```bash
curl -X 'GET' \
'https://taskmanagerapi-ttkjqk.backend.im/tasks/1' \
-H 'accept: application/json'
-H 'accept: application/json' \
-H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
```
### Update a task
### Update a task (with authentication)
```bash
curl -X 'PUT' \
'https://taskmanagerapi-ttkjqk.backend.im/tasks/1' \
-H 'accept: application/json' \
-H 'Authorization: Bearer YOUR_ACCESS_TOKEN' \
-H 'Content-Type: application/json' \
-d '{
"title": "Updated title",
@ -88,20 +128,31 @@ curl -X 'PUT' \
}'
```
### Delete a task
### Delete a task (with authentication)
```bash
curl -X 'DELETE' \
'https://taskmanagerapi-ttkjqk.backend.im/tasks/1' \
-H 'accept: application/json'
-H 'accept: application/json' \
-H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
```
### Mark a task as completed
### Mark a task as completed (with authentication)
```bash
curl -X 'POST' \
'https://taskmanagerapi-ttkjqk.backend.im/tasks/1/complete' \
-H 'accept: application/json'
-H 'accept: application/json' \
-H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
```
### Get current user information (with authentication)
```bash
curl -X 'GET' \
'https://taskmanagerapi-ttkjqk.backend.im/auth/me' \
-H 'accept: application/json' \
-H 'Authorization: Bearer YOUR_ACCESS_TOKEN'
```
## Project Structure
@ -112,12 +163,28 @@ taskmanagerapi/
│ └── versions/ # Migration scripts
├── app/
│ ├── api/ # API endpoints
│ │ ├── deps.py # Dependency injection for authentication
│ │ └── routers/ # API route definitions
│ │ ├── auth.py # Authentication endpoints
│ │ └── tasks.py # Task management endpoints
│ ├── core/ # Core application code
│ │ ├── config.py # Application configuration
│ │ └── security.py # Security utilities (JWT, password hashing)
│ ├── crud/ # CRUD operations
│ ├── db/ # Database setup and models
│ │ ├── base.py # Base CRUD operations
│ │ ├── task.py # Task CRUD operations
│ │ └── user.py # User CRUD operations
│ ├── db/ # Database setup
│ │ ├── base.py # Base imports for models
│ │ ├── base_class.py # Base class for SQLAlchemy models
│ │ ├── init_db.py # Database initialization
│ │ └── session.py # Database session management
│ ├── models/ # SQLAlchemy models
│ │ ├── task.py # Task model
│ │ └── user.py # User model
│ └── schemas/ # Pydantic schemas/models
│ ├── task.py # Task schemas
│ └── user.py # User and authentication schemas
├── main.py # Application entry point
└── requirements.txt # Project dependencies
```
@ -183,6 +250,22 @@ uvicorn main:app --reload
The API will be available at http://localhost:8000
### Default Admin User
During initialization, the system creates a default admin user:
- Email: admin@example.com
- Username: admin
- Password: adminpassword
You can use these credentials to authenticate and get started right away.
### Authentication Flow
1. **Register a new user**: Send a POST request to `/auth/register` with email, username, and password
2. **Login**: Send a POST request to `/auth/login` with username/email and password to receive an access token
3. **Use token**: Include the token in your requests as a Bearer token in the Authorization header
4. **Access protected endpoints**: Use the token to access protected task management endpoints
### API Documentation
- Swagger UI: http://localhost:8000/docs

57
alembic/env.py
View File

@ -11,7 +11,7 @@ from sqlalchemy import pool
from alembic import context
# Add the parent directory to the Python path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
@ -29,27 +29,27 @@ db_url = config.get_main_option("sqlalchemy.url")
if db_url.startswith("sqlite:///"):
# Extract the path part after sqlite:///
if db_url.startswith("sqlite:////"): # Absolute path (4 slashes)
db_path = db_url[len("sqlite:///"):]
db_path = db_url[len("sqlite:///") :]
else: # Relative path (3 slashes)
db_path = db_url[len("sqlite:///"):]
db_path = db_url[len("sqlite:///") :]
# Get the directory path
db_dir = os.path.dirname(db_path)
logger.info(f"Database URL: {db_url}")
logger.info(f"Database path: {db_path}")
logger.info(f"Database directory: {db_dir}")
# Create directory if it doesn't exist
try:
os.makedirs(db_dir, exist_ok=True)
logger.info(f"Ensured database directory exists: {db_dir}")
# Test if we can create the database file
try:
# Try to touch the database file
Path(db_path).touch(exist_ok=True)
logger.info(f"Database file is accessible: {db_path}")
# Test direct SQLite connection
try:
conn = sqlite3.connect(db_path)
@ -66,6 +66,7 @@ if db_url.startswith("sqlite:///"):
# add your model's MetaData object here
# for 'autogenerate' support
from app.db.base import Base # noqa
target_metadata = Base.metadata
# other values from the config, defined by the needs of env.py,
@ -89,7 +90,7 @@ def run_migrations_offline():
try:
url = config.get_main_option("sqlalchemy.url")
logger.info(f"Running offline migrations using URL: {url}")
context.configure(
url=url,
target_metadata=target_metadata,
@ -101,7 +102,7 @@ def run_migrations_offline():
logger.info("Running offline migrations...")
context.run_migrations()
logger.info("Offline migrations completed successfully")
except Exception as e:
logger.error(f"Offline migration error: {e}")
# Re-raise the error
@ -120,11 +121,11 @@ def run_migrations_online():
cfg = config.get_section(config.config_ini_section)
url = cfg.get("sqlalchemy.url")
logger.info(f"Running online migrations using URL: {url}")
# Create engine with retry logic
max_retries = 3
last_error = None
for retry in range(max_retries):
try:
logger.info(f"Connection attempt {retry + 1}/{max_retries}")
@ -133,13 +134,13 @@ def run_migrations_online():
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
# Configure SQLite for better reliability
@event.listens_for(connectable, "connect")
def setup_sqlite_connection(dbapi_connection, connection_record):
dbapi_connection.execute("PRAGMA journal_mode=WAL")
dbapi_connection.execute("PRAGMA synchronous=NORMAL")
# Connect and run migrations
with connectable.connect() as connection:
logger.info("Connection successful")
@ -152,35 +153,39 @@ def run_migrations_online():
context.run_migrations()
logger.info("Migrations completed successfully")
return # Success, exit the function
except Exception as e:
last_error = e
logger.error(f"Migration attempt {retry + 1} failed: {e}")
if retry < max_retries - 1:
import time
wait_time = (retry + 1) * 2 # Exponential backoff
logger.info(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
# If we get here, all retries failed
raise Exception(f"Failed to run migrations after {max_retries} attempts: {last_error}")
raise Exception(
f"Failed to run migrations after {max_retries} attempts: {last_error}"
)
except Exception as e:
logger.error(f"Migration error: {e}")
# Print diagnostic information
from sqlalchemy import __version__ as sa_version
logger.error(f"SQLAlchemy version: {sa_version}")
# Get directory info
if url and url.startswith("sqlite:///"):
if url.startswith("sqlite:////"): # Absolute path
db_path = url[len("sqlite:///"):]
db_path = url[len("sqlite:///") :]
else: # Relative path
db_path = url[len("sqlite:///"):]
db_path = url[len("sqlite:///") :]
db_dir = os.path.dirname(db_path)
# Directory permissions
if os.path.exists(db_dir):
stats = os.stat(db_dir)
@ -189,7 +194,7 @@ def run_migrations_online():
logger.error(f"DB directory is writable: {os.access(db_dir, os.W_OK)}")
else:
logger.error("DB directory exists: No")
# File permissions if file exists
if os.path.exists(db_path):
stats = os.stat(db_path)
@ -198,7 +203,7 @@ def run_migrations_online():
logger.error(f"DB file is writable: {os.access(db_path, os.W_OK)}")
else:
logger.error("DB file exists: No")
# Re-raise the error
raise
@ -206,4 +211,4 @@ def run_migrations_online():
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
run_migrations_online()

44
alembic/versions/0001_create_tasks_table.py
View File

@ -1,16 +1,16 @@
"""create tasks table
Revision ID: 0001
Revises:
Revises:
Create Date: 2025-05-14
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import sqlite
# revision identifiers, used by Alembic.
revision = '0001'
revision = "0001"
down_revision = None
branch_labels = None
depends_on = None
@ -18,21 +18,31 @@ depends_on = None
def upgrade():
op.create_table(
'task',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(length=100), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('priority', sa.Enum('low', 'medium', 'high', name='taskpriority'), default='medium'),
sa.Column('status', sa.Enum('todo', 'in_progress', 'done', name='taskstatus'), default='todo'),
sa.Column('due_date', sa.DateTime(), nullable=True),
sa.Column('completed', sa.Boolean(), default=False),
sa.Column('created_at', sa.DateTime(), default=sa.func.now()),
sa.Column('updated_at', sa.DateTime(), default=sa.func.now(), onupdate=sa.func.now()),
sa.PrimaryKeyConstraint('id')
"task",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("title", sa.String(length=100), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column(
"priority",
sa.Enum("low", "medium", "high", name="taskpriority"),
default="medium",
),
sa.Column(
"status",
sa.Enum("todo", "in_progress", "done", name="taskstatus"),
default="todo",
),
sa.Column("due_date", sa.DateTime(), nullable=True),
sa.Column("completed", sa.Boolean(), default=False),
sa.Column("created_at", sa.DateTime(), default=sa.func.now()),
sa.Column(
"updated_at", sa.DateTime(), default=sa.func.now(), onupdate=sa.func.now()
),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f('ix_task_id'), 'task', ['id'], unique=False)
op.create_index(op.f("ix_task_id"), "task", ["id"], unique=False)
def downgrade():
op.drop_index(op.f('ix_task_id'), table_name='task')
op.drop_table('task')
op.drop_index(op.f("ix_task_id"), table_name="task")
op.drop_table("task")

58
alembic/versions/0002_add_user_table.py Normal file
View File

@ -0,0 +1,58 @@
"""Add user table and task relation
Revision ID: 0002
Revises: 0001_create_tasks_table
Create Date: 2023-10-31
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic
revision = "0002"
down_revision = "0001_create_tasks_table"
branch_labels = None
depends_on = None
def upgrade():
# Create user table
op.create_table(
"user",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("email", sa.String(length=255), nullable=False),
sa.Column("username", sa.String(length=100), nullable=False),
sa.Column("hashed_password", sa.String(length=255), nullable=False),
sa.Column("is_active", sa.Boolean(), nullable=True, default=True),
sa.Column("is_superuser", sa.Boolean(), nullable=True, default=False),
sa.Column("created_at", sa.DateTime(), nullable=True),
sa.Column("updated_at", sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint("id"),
)
# Create indexes on user table
op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
op.create_index(op.f("ix_user_id"), "user", ["id"], unique=False)
op.create_index(op.f("ix_user_username"), "user", ["username"], unique=True)
# Add user_id column to task table
with op.batch_alter_table("task", schema=None) as batch_op:
batch_op.add_column(sa.Column("user_id", sa.Integer(), nullable=True))
batch_op.create_foreign_key("fk_task_user_id", "user", ["user_id"], ["id"])
def downgrade():
# Remove foreign key and user_id column from task table
with op.batch_alter_table("task", schema=None) as batch_op:
batch_op.drop_constraint("fk_task_user_id", type_="foreignkey")
batch_op.drop_column("user_id")
# Drop indexes on user table
op.drop_index(op.f("ix_user_username"), table_name="user")
op.drop_index(op.f("ix_user_id"), table_name="user")
op.drop_index(op.f("ix_user_email"), table_name="user")
# Drop user table
op.drop_table("user")

2
app/__init__.py
View File

@ -1 +1 @@
# App package
# App package

2
app/api/__init__.py
View File

@ -1 +1 @@
# API package
# API package

63
app/api/deps.py
View File

@ -1,4 +1,61 @@
from typing import Generator
from typing import Optional
# Use the improved get_db function from session.py
from app.db.session import get_db
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from pydantic import ValidationError
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.core.config import settings
from app.core.security import ALGORITHM
from app.models.user import User
from app.schemas.user import TokenPayload
from app import crud
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login", auto_error=False)
def get_current_user(
db: Session = Depends(get_db), token: Optional[str] = Depends(oauth2_scheme)
) -> Optional[User]:
"""
Get the current user from the provided JWT token.
Returns None if no token is provided or the token is invalid.
"""
if not token:
return None
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
token_data = TokenPayload(**payload)
except (JWTError, ValidationError):
return None
user = crud.user.get(db, id=token_data.sub)
if not user:
return None
return user
def get_current_active_user(
current_user: User = Depends(get_current_user),
) -> User:
"""
Get the current active user and raise an exception if the user is not
authenticated or inactive.
"""
if not current_user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
if not crud.user.is_active(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user"
)
return current_user

5
app/api/routers/__init__.py
View File

@ -1,6 +1,7 @@
from fastapi import APIRouter
from app.api.routers import tasks
from app.api.routers import tasks, auth
api_router = APIRouter()
api_router.include_router(tasks.router, prefix="/tasks", tags=["tasks"])
api_router.include_router(tasks.router, prefix="/tasks", tags=["tasks"])
api_router.include_router(auth.router, prefix="/auth", tags=["auth"])

95
app/api/routers/auth.py Normal file
View File

@ -0,0 +1,95 @@
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app import crud
from app.api.deps import get_db, get_current_active_user
from app.core.config import settings
from app.core.security import create_access_token
from app.models.user import User
from app.schemas.user import User as UserSchema
from app.schemas.user import UserCreate, Token
router = APIRouter()
@router.post("/register", response_model=UserSchema)
def register(
*,
db: Session = Depends(get_db),
user_in: UserCreate,
) -> Any:
"""
Register a new user.
"""
# Check if user with this email already exists
user = crud.user.get_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="A user with this email already exists.",
)
# Check if user with this username already exists
user = crud.user.get_by_username(db, username=user_in.username)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="A user with this username already exists.",
)
# Create new user
user = crud.user.create(db, obj_in=user_in)
return user
@router.post("/login", response_model=Token)
def login(
db: Session = Depends(get_db),
form_data: OAuth2PasswordRequestForm = Depends(),
) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests.
"""
user = crud.user.authenticate(
db, email_or_username=form_data.username, password=form_data.password
)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email/username or password",
headers={"WWW-Authenticate": "Bearer"},
)
if not crud.user.is_active(user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Inactive user",
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
subject=user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/test-token", response_model=UserSchema)
def test_token(current_user: User = Depends(get_current_active_user)) -> Any:
"""
Test access token.
"""
return current_user
@router.get("/me", response_model=UserSchema)
def read_users_me(current_user: User = Depends(get_current_active_user)) -> Any:
"""
Get current user.
"""
return current_user

451
app/api/routers/tasks.py
View File

@ -1,11 +1,12 @@
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app import crud
from app.api.deps import get_db
from app.api.deps import get_db, get_current_active_user
from app.models.task import TaskStatus
from app.models.user import User
from app.schemas.task import Task, TaskCreate, TaskUpdate
router = APIRouter()
@ -17,73 +18,82 @@ def read_tasks(
skip: int = 0,
limit: int = 100,
status: Optional[TaskStatus] = None,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve tasks.
Retrieve tasks for the current user.
"""
try:
import traceback
import sqlite3
from sqlalchemy import text
from app.db.session import db_file
print(f"Getting tasks with status: {status}, skip: {skip}, limit: {limit}")
# Try the normal SQLAlchemy approach first
try:
if status:
tasks = crud.task.get_by_status(db, status=status)
tasks = crud.task.get_by_status(
db, status=status, user_id=current_user.id
)
else:
tasks = crud.task.get_multi(db, skip=skip, limit=limit)
tasks = crud.task.get_multi(
db, skip=skip, limit=limit, user_id=current_user.id
)
return tasks
except Exception as e:
print(f"Error getting tasks with SQLAlchemy: {e}")
print(traceback.format_exc())
# Continue to fallback
# Fallback to direct SQLite approach
try:
conn = sqlite3.connect(str(db_file))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
if status:
cursor.execute("SELECT * FROM task WHERE status = ? LIMIT ? OFFSET ?",
(status.value, limit, skip))
cursor.execute(
"SELECT * FROM task WHERE status = ? AND user_id = ? LIMIT ? OFFSET ?",
(status.value, current_user.id, limit, skip),
)
else:
cursor.execute("SELECT * FROM task LIMIT ? OFFSET ?", (limit, skip))
cursor.execute(
"SELECT * FROM task WHERE user_id = ? LIMIT ? OFFSET ?",
(current_user.id, limit, skip),
)
rows = cursor.fetchall()
# Convert to Task objects
tasks = []
for row in rows:
task_dict = dict(row)
# Convert completed to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
if "completed" in task_dict:
task_dict["completed"] = bool(task_dict["completed"])
# Convert to object with attributes
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
tasks.append(TaskResult(**task_dict))
conn.close()
return tasks
except Exception as e:
print(f"Error getting tasks with direct SQLite: {e}")
print(traceback.format_exc())
raise
except Exception as e:
print(f"Global error in read_tasks: {e}")
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error retrieving tasks: {str(e)}"
detail=f"Error retrieving tasks: {str(e)}",
)
@ -92,9 +102,10 @@ def create_task(
*,
db: Session = Depends(get_db),
task_in: TaskCreate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Create new task - using direct SQLite approach for reliability.
Create new task for the current user - using direct SQLite approach for reliability.
"""
import sqlite3
import time
@ -102,60 +113,60 @@ def create_task(
import traceback
from datetime import datetime
from app.db.session import db_file
# Log creation attempt
print(f"[{datetime.now().isoformat()}] Task creation requested", file=sys.stdout)
# Use direct SQLite for maximum reliability
try:
# Extract task data regardless of Pydantic version
try:
if hasattr(task_in, 'model_dump'):
if hasattr(task_in, "model_dump"):
task_data = task_in.model_dump()
elif hasattr(task_in, 'dict'):
elif hasattr(task_in, "dict"):
task_data = task_in.dict()
else:
# Fallback for any case
task_data = {
'title': getattr(task_in, 'title', 'Untitled Task'),
'description': getattr(task_in, 'description', ''),
'priority': getattr(task_in, 'priority', 'medium'),
'status': getattr(task_in, 'status', 'todo'),
'due_date': getattr(task_in, 'due_date', None),
'completed': getattr(task_in, 'completed', False)
"title": getattr(task_in, "title", "Untitled Task"),
"description": getattr(task_in, "description", ""),
"priority": getattr(task_in, "priority", "medium"),
"status": getattr(task_in, "status", "todo"),
"due_date": getattr(task_in, "due_date", None),
"completed": getattr(task_in, "completed", False),
}
print(f"Task data: {task_data}")
except Exception as e:
print(f"Error extracting task data: {e}")
# Fallback to minimal data
task_data = {
'title': str(getattr(task_in, 'title', 'Unknown Title')),
'description': str(getattr(task_in, 'description', '')),
'priority': 'medium',
'status': 'todo',
'completed': False
"title": str(getattr(task_in, "title", "Unknown Title")),
"description": str(getattr(task_in, "description", "")),
"priority": "medium",
"status": "todo",
"completed": False,
}
# Format due_date if present
if task_data.get('due_date'):
if task_data.get("due_date"):
try:
if isinstance(task_data['due_date'], datetime):
task_data['due_date'] = task_data['due_date'].isoformat()
elif isinstance(task_data['due_date'], str):
if isinstance(task_data["due_date"], datetime):
task_data["due_date"] = task_data["due_date"].isoformat()
elif isinstance(task_data["due_date"], str):
# Standardize format by parsing and reformatting
parsed_date = datetime.fromisoformat(
task_data['due_date'].replace('Z', '+00:00')
task_data["due_date"].replace("Z", "+00:00")
)
task_data['due_date'] = parsed_date.isoformat()
task_data["due_date"] = parsed_date.isoformat()
except Exception as e:
print(f"Warning: Could not parse due_date: {e}")
# Keep as-is or set to None if invalid
if not isinstance(task_data['due_date'], str):
task_data['due_date'] = None
if not isinstance(task_data["due_date"], str):
task_data["due_date"] = None
# Get current timestamp for created/updated fields
now = datetime.utcnow().isoformat()
# Connect to SQLite with retry logic
for retry in range(3):
conn = None
@ -163,7 +174,7 @@ def create_task(
# Try to connect to the database with a timeout
conn = sqlite3.connect(str(db_file), timeout=30)
cursor = conn.cursor()
# Create the task table if it doesn't exist - using minimal schema
cursor.execute("""
CREATE TABLE IF NOT EXISTS task (
@ -175,122 +186,134 @@ def create_task(
due_date TEXT,
completed INTEGER DEFAULT 0,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
user_id INTEGER
)
""")
# Insert the task - provide defaults for all fields
cursor.execute(
"""
INSERT INTO task (
title, description, priority, status,
due_date, completed, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
due_date, completed, created_at, updated_at, user_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
task_data.get('title', 'Untitled'),
task_data.get('description', ''),
task_data.get('priority', 'medium'),
task_data.get('status', 'todo'),
task_data.get('due_date'),
1 if task_data.get('completed') else 0,
task_data.get("title", "Untitled"),
task_data.get("description", ""),
task_data.get("priority", "medium"),
task_data.get("status", "todo"),
task_data.get("due_date"),
1 if task_data.get("completed") else 0,
now,
now
)
now,
current_user.id,
),
)
# Get the ID of the inserted task
task_id = cursor.lastrowid
print(f"Task inserted with ID: {task_id}")
# Commit the transaction
conn.commit()
# Retrieve the created task to return it
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
cursor.execute(
"SELECT * FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
row = cursor.fetchone()
if row:
# Get column names from cursor description
column_names = [desc[0] for desc in cursor.description]
# Create a dictionary from row values
task_dict = dict(zip(column_names, row))
# Convert 'completed' to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
if "completed" in task_dict:
task_dict["completed"] = bool(task_dict["completed"])
# Create an object that mimics the Task model
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
print(f"Task created successfully: ID={task_id}")
# Close the connection and return the task
conn.close()
return TaskResult(**task_dict)
else:
conn.close()
raise Exception(f"Task creation succeeded but retrieval failed for ID: {task_id}")
raise Exception(
f"Task creation succeeded but retrieval failed for ID: {task_id}"
)
except sqlite3.OperationalError as e:
if conn:
conn.close()
# Check if retry is appropriate
if "database is locked" in str(e) and retry < 2:
wait_time = (retry + 1) * 1.5 # Exponential backoff
print(f"Database locked, retrying in {wait_time}s (attempt {retry+1}/3)")
print(
f"Database locked, retrying in {wait_time}s (attempt {retry + 1}/3)"
)
time.sleep(wait_time)
else:
print(f"SQLite operational error: {e}")
raise
except Exception as e:
if conn:
# Try to rollback if connection is still open
try:
conn.rollback()
except:
except Exception:
pass
conn.close()
print(f"Error in SQLite task creation: {e}")
print(traceback.format_exc())
# Only retry on specific transient errors
if retry < 2 and ("locked" in str(e).lower() or "busy" in str(e).lower()):
if retry < 2 and (
"locked" in str(e).lower() or "busy" in str(e).lower()
):
time.sleep(1)
continue
raise
# If we reach here, the retry loop failed
raise Exception("Failed to create task after multiple attempts")
except Exception as sqlite_error:
# Final fallback: try SQLAlchemy approach
try:
print(f"Direct SQLite approach failed: {sqlite_error}")
print("Trying SQLAlchemy as fallback...")
task = crud.task.create(db, obj_in=task_in)
task = crud.task.create_with_owner(
db, obj_in=task_in, user_id=current_user.id
)
print(f"Task created with SQLAlchemy fallback: ID={task.id}")
return task
except Exception as alch_error:
print(f"SQLAlchemy fallback also failed: {alch_error}")
print(traceback.format_exc())
# Provide detailed error information
error_detail = f"Task creation failed. Primary error: {str(sqlite_error)}. Fallback error: {str(alch_error)}"
print(f"Final error: {error_detail}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=error_detail
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=error_detail
)
@ -299,54 +322,60 @@ def read_task(
*,
db: Session = Depends(get_db),
task_id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Get task by ID.
Get task by ID for the current user.
"""
try:
import traceback
import sqlite3
from app.db.session import db_file
print(f"Getting task with ID: {task_id}")
# Try the normal SQLAlchemy approach first
try:
task = crud.task.get(db, id=task_id)
task = crud.task.get_by_id_and_user(
db, task_id=task_id, user_id=current_user.id
)
if task:
return task
# Fall through to direct SQLite if task not found
# Task not found or doesn't belong to user - check further with direct SQLite
except Exception as e:
print(f"Error getting task with SQLAlchemy: {e}")
print(traceback.format_exc())
# Continue to fallback
# Fallback to direct SQLite approach
try:
conn = sqlite3.connect(str(db_file))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
cursor.execute(
"SELECT * FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
row = cursor.fetchone()
if not row:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
task_dict = dict(row)
# Convert completed to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
if "completed" in task_dict:
task_dict["completed"] = bool(task_dict["completed"])
# Convert to object with attributes
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
conn.close()
return TaskResult(**task_dict)
except HTTPException:
@ -356,9 +385,9 @@ def read_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error retrieving task: {str(e)}"
detail=f"Error retrieving task: {str(e)}",
)
except HTTPException:
raise # Re-raise any HTTP exceptions
except Exception as e:
@ -366,7 +395,7 @@ def read_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error retrieving task: {str(e)}"
detail=f"Error retrieving task: {str(e)}",
)
@ -376,30 +405,34 @@ def update_task(
db: Session = Depends(get_db),
task_id: int,
task_in: TaskUpdate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Update a task.
Update a task for the current user.
"""
try:
import traceback
import sqlite3
import json
from datetime import datetime
from app.db.session import db_file
print(f"Updating task with ID: {task_id}, data: {task_in}")
# Handle datetime conversion for due_date if present
if hasattr(task_in, "due_date") and task_in.due_date is not None:
if isinstance(task_in.due_date, str):
try:
task_in.due_date = datetime.fromisoformat(task_in.due_date.replace('Z', '+00:00'))
task_in.due_date = datetime.fromisoformat(
task_in.due_date.replace("Z", "+00:00")
)
except Exception as e:
print(f"Error parsing due_date: {e}")
# Try the normal SQLAlchemy approach first
try:
task = crud.task.get(db, id=task_id)
task = crud.task.get_by_id_and_user(
db, task_id=task_id, user_id=current_user.id
)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
@ -413,84 +446,94 @@ def update_task(
print(f"Error updating task with SQLAlchemy: {e}")
print(traceback.format_exc())
# Continue to fallback
# Fallback to direct SQLite approach
try:
conn = sqlite3.connect(str(db_file))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# First check if task exists
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
# First check if task exists and belongs to current user
cursor.execute(
"SELECT * FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
row = cursor.fetchone()
if not row:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
# Convert Pydantic model to dict, excluding unset values
updates = {}
model_data = task_in.model_dump(exclude_unset=True) if hasattr(task_in, "model_dump") else task_in.dict(exclude_unset=True)
model_data = (
task_in.model_dump(exclude_unset=True)
if hasattr(task_in, "model_dump")
else task_in.dict(exclude_unset=True)
)
# Only include fields that were provided in the update
for key, value in model_data.items():
if value is not None: # Skip None values
updates[key] = value
if not updates:
# No updates provided
task_dict = dict(row)
# Convert completed to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
if "completed" in task_dict:
task_dict["completed"] = bool(task_dict["completed"])
# Return the unchanged task
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
conn.close()
return TaskResult(**task_dict)
# Format datetime objects
if 'due_date' in updates and isinstance(updates['due_date'], datetime):
updates['due_date'] = updates['due_date'].isoformat()
if "due_date" in updates and isinstance(updates["due_date"], datetime):
updates["due_date"] = updates["due_date"].isoformat()
# Add updated_at timestamp
updates['updated_at'] = datetime.utcnow().isoformat()
updates["updated_at"] = datetime.utcnow().isoformat()
# Build the SQL update statement
set_clause = ", ".join([f"{key} = ?" for key in updates.keys()])
params = list(updates.values())
params.append(task_id) # For the WHERE clause
cursor.execute(f"UPDATE task SET {set_clause} WHERE id = ?", params)
conn.commit()
# Return the updated task
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
cursor.execute(
"SELECT * FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
updated_row = cursor.fetchone()
conn.close()
if updated_row:
task_dict = dict(updated_row)
# Convert completed to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
if "completed" in task_dict:
task_dict["completed"] = bool(task_dict["completed"])
# Convert to object with attributes
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
return TaskResult(**task_dict)
else:
raise Exception("Task was updated but could not be retrieved")
except HTTPException:
raise # Re-raise the 404 exception
except Exception as e:
@ -498,9 +541,9 @@ def update_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error updating task: {str(e)}"
detail=f"Error updating task: {str(e)}",
)
except HTTPException:
raise # Re-raise any HTTP exceptions
except Exception as e:
@ -508,7 +551,7 @@ def update_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error updating task: {str(e)}"
detail=f"Error updating task: {str(e)}",
)
@ -517,72 +560,77 @@ def delete_task(
*,
db: Session = Depends(get_db),
task_id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Delete a task.
Delete a task for the current user.
"""
try:
import traceback
import sqlite3
from app.db.session import db_file
print(f"Deleting task with ID: {task_id}")
# First, get the task to return it later
task_to_return = None
# Try the normal SQLAlchemy approach first
# Try the normal SQLAlchemy approach
try:
task = crud.task.get(db, id=task_id)
task = crud.task.get_by_id_and_user(
db, task_id=task_id, user_id=current_user.id
)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
task_to_return = task
task = crud.task.remove(db, id=task_id)
return task
removed_task = crud.task.remove(db, id=task_id)
return removed_task
except HTTPException:
raise # Re-raise the 404 exception
except Exception as e:
print(f"Error deleting task with SQLAlchemy: {e}")
print(traceback.format_exc())
# Continue to fallback
# Fallback to direct SQLite approach
try:
conn = sqlite3.connect(str(db_file))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# First save the task data for the return value
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
cursor.execute(
"SELECT * FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
row = cursor.fetchone()
if not row:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
task_dict = dict(row)
# Convert completed to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
if "completed" in task_dict:
task_dict["completed"] = bool(task_dict["completed"])
# Delete the task
cursor.execute("DELETE FROM task WHERE id = ?", (task_id,))
cursor.execute(
"DELETE FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
conn.commit()
conn.close()
# Convert to object with attributes
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
return TaskResult(**task_dict)
except HTTPException:
raise # Re-raise the 404 exception
except Exception as e:
@ -590,9 +638,9 @@ def delete_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error deleting task: {str(e)}"
detail=f"Error deleting task: {str(e)}",
)
except HTTPException:
raise # Re-raise any HTTP exceptions
except Exception as e:
@ -600,7 +648,7 @@ def delete_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error deleting task: {str(e)}"
detail=f"Error deleting task: {str(e)}",
)
@ -609,21 +657,24 @@ def complete_task(
*,
db: Session = Depends(get_db),
task_id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Mark a task as completed.
Mark a task as completed for the current user.
"""
try:
import traceback
import sqlite3
from datetime import datetime
from app.db.session import db_file
print(f"Marking task {task_id} as completed")
# Try the normal SQLAlchemy approach first
try:
task = crud.task.mark_completed(db, task_id=task_id)
task = crud.task.mark_completed(
db, task_id=task_id, user_id=current_user.id
)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
@ -634,52 +685,58 @@ def complete_task(
print(f"Error completing task with SQLAlchemy: {e}")
print(traceback.format_exc())
# Continue to fallback
# Fallback to direct SQLite approach
try:
conn = sqlite3.connect(str(db_file))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# First check if task exists
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
# First check if task exists and belongs to current user
cursor.execute(
"SELECT * FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
row = cursor.fetchone()
if not row:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Task not found",
)
# Update task to completed status
now = datetime.utcnow().isoformat()
cursor.execute(
"UPDATE task SET completed = ?, status = ?, updated_at = ? WHERE id = ?",
(1, "done", now, task_id)
"UPDATE task SET completed = ?, status = ?, updated_at = ? WHERE id = ? AND user_id = ?",
(1, "done", now, task_id, current_user.id),
)
conn.commit()
# Get the updated task
cursor.execute("SELECT * FROM task WHERE id = ?", (task_id,))
cursor.execute(
"SELECT * FROM task WHERE id = ? AND user_id = ?",
(task_id, current_user.id),
)
updated_row = cursor.fetchone()
conn.close()
if updated_row:
task_dict = dict(updated_row)
# Convert completed to boolean
if 'completed' in task_dict:
task_dict['completed'] = bool(task_dict['completed'])
if "completed" in task_dict:
task_dict["completed"] = bool(task_dict["completed"])
# Convert to object with attributes
class TaskResult:
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
return TaskResult(**task_dict)
else:
raise Exception("Task was completed but could not be retrieved")
except HTTPException:
raise # Re-raise the 404 exception
except Exception as e:
@ -687,9 +744,9 @@ def complete_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error completing task: {str(e)}"
detail=f"Error completing task: {str(e)}",
)
except HTTPException:
raise # Re-raise any HTTP exceptions
except Exception as e:
@ -697,5 +754,5 @@ def complete_task(
print(traceback.format_exc())
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Error completing task: {str(e)}"
)
detail=f"Error completing task: {str(e)}",
)

25
app/core/config.py
View File

@ -1,6 +1,6 @@
import secrets
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from typing import List, Union
from pydantic import AnyHttpUrl, field_validator
from pydantic_settings import BaseSettings
@ -16,11 +16,11 @@ if DB_PATH:
else:
# Try production path first, then local directory
paths_to_try = [
Path("/app/db"), # Production path
Path.cwd() / "db", # Local development path
Path("/tmp/taskmanager") # Fallback path
Path("/app/db"), # Production path
Path.cwd() / "db", # Local development path
Path("/tmp/taskmanager"), # Fallback path
]
# Find the first writable path
DB_DIR = None
for path in paths_to_try:
@ -35,16 +35,17 @@ else:
break
except Exception as e:
print(f"Cannot use path {path}: {e}")
# Last resort fallback
if DB_DIR is None:
DB_DIR = Path("/tmp")
print(f"Falling back to temporary directory: {DB_DIR}")
try:
Path("/tmp").mkdir(exist_ok=True)
except:
except Exception:
pass
class Settings(BaseSettings):
PROJECT_NAME: str = "Task Manager API"
# No API version prefix - use direct paths
@ -52,7 +53,7 @@ class Settings(BaseSettings):
SECRET_KEY: str = secrets.token_urlsafe(32)
# 60 minutes * 24 hours * 8 days = 8 days
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
# CORS Configuration
BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
@ -66,10 +67,8 @@ class Settings(BaseSettings):
# Database configuration
SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
model_config = {
"case_sensitive": True
}
model_config = {"case_sensitive": True}
settings = Settings()
settings = Settings()

42
app/core/security.py Normal file
View File

@ -0,0 +1,42 @@
from datetime import datetime, timedelta
from typing import Any, Optional, Union
from jose import jwt
from passlib.context import CryptContext
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
ALGORITHM = "HS256"
def create_access_token(
subject: Union[str, Any], expires_delta: Optional[timedelta] = None
) -> str:
"""
Create a JWT access token for a user
"""
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""
Verify a password against a hash
"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""
Hash a password
"""
return pwd_context.hash(password)

5
app/crud/__init__.py
View File

@ -1 +1,4 @@
from app.crud.task import task
from app.crud.task import task as task
from app.crud.user import user as user
__all__ = ["task", "user"]

33
app/crud/base.py
View File

@ -30,12 +30,12 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
try:
obj_in_data = jsonable_encoder(obj_in)
print(f"Creating {self.model.__name__} with data: {obj_in_data}")
db_obj = self.model(**obj_in_data)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
print(f"Successfully created {self.model.__name__} with id: {db_obj.id}")
return db_obj
except Exception as e:
@ -43,6 +43,7 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
error_msg = f"Error creating {self.model.__name__}: {str(e)}"
print(error_msg)
import traceback
print(traceback.format_exc())
raise Exception(error_msg) from e
@ -51,15 +52,15 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
db: Session,
*,
db_obj: ModelType,
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
obj_in: Union[UpdateSchemaType, Dict[str, Any]],
) -> ModelType:
try:
# Log update operation
print(f"Updating {self.model.__name__} with id: {db_obj.id}")
# Get the existing data
obj_data = jsonable_encoder(db_obj)
# Process the update data
if isinstance(obj_in, dict):
update_data = obj_in
@ -69,29 +70,30 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
update_data = obj_in.model_dump(exclude_unset=True)
else:
update_data = obj_in.dict(exclude_unset=True)
# Log the changes being made
changes = {k: v for k, v in update_data.items() if k in obj_data}
print(f"Fields to update: {changes}")
# Apply the updates
for field in obj_data:
if field in update_data:
setattr(db_obj, field, update_data[field])
# Save changes
db.add(db_obj)
db.commit()
db.refresh(db_obj)
print(f"Successfully updated {self.model.__name__} with id: {db_obj.id}")
return db_obj
except Exception as e:
db.rollback()
error_msg = f"Error updating {self.model.__name__}: {str(e)}"
print(error_msg)
import traceback
print(traceback.format_exc())
raise Exception(error_msg) from e
@ -102,20 +104,21 @@ class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
if not obj:
print(f"{self.model.__name__} with id {id} not found for deletion")
return None
print(f"Deleting {self.model.__name__} with id: {id}")
# Delete the object
db.delete(obj)
db.commit()
print(f"Successfully deleted {self.model.__name__} with id: {id}")
return obj
except Exception as e:
db.rollback()
error_msg = f"Error deleting {self.model.__name__}: {str(e)}"
print(error_msg)
import traceback
print(traceback.format_exc())
raise Exception(error_msg) from e
raise Exception(error_msg) from e

74
app/crud/task.py
View File

@ -7,21 +7,69 @@ from app.schemas.task import TaskCreate, TaskUpdate
class CRUDTask(CRUDBase[Task, TaskCreate, TaskUpdate]):
def get_by_status(self, db: Session, *, status: TaskStatus) -> List[Task]:
return db.query(self.model).filter(Task.status == status).all()
def get_completed(self, db: Session) -> List[Task]:
return db.query(self.model).filter(Task.completed == True).all()
def mark_completed(self, db: Session, *, task_id: int) -> Optional[Task]:
task = self.get(db, id=task_id)
def get_by_status(
self, db: Session, *, status: TaskStatus, user_id: Optional[int] = None
) -> List[Task]:
query = db.query(self.model).filter(Task.status == status)
if user_id is not None:
query = query.filter(Task.user_id == user_id)
return query.all()
def get_completed(
self, db: Session, *, user_id: Optional[int] = None
) -> List[Task]:
query = db.query(self.model).filter(Task.completed.is_(True))
if user_id is not None:
query = query.filter(Task.user_id == user_id)
return query.all()
def get_multi(
self,
db: Session,
*,
skip: int = 0,
limit: int = 100,
user_id: Optional[int] = None,
) -> List[Task]:
query = db.query(self.model)
if user_id is not None:
query = query.filter(Task.user_id == user_id)
return query.offset(skip).limit(limit).all()
def create_with_owner(
self, db: Session, *, obj_in: TaskCreate, user_id: int
) -> Task:
obj_in_data = (
obj_in.model_dump() if hasattr(obj_in, "model_dump") else obj_in.dict()
)
db_obj = self.model(**obj_in_data, user_id=user_id)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def get_by_id_and_user(
self, db: Session, *, task_id: int, user_id: int
) -> Optional[Task]:
return (
db.query(self.model)
.filter(Task.id == task_id, Task.user_id == user_id)
.first()
)
def mark_completed(
self, db: Session, *, task_id: int, user_id: Optional[int] = None
) -> Optional[Task]:
if user_id:
task = self.get_by_id_and_user(db, task_id=task_id, user_id=user_id)
else:
task = self.get(db, id=task_id)
if not task:
return None
task_in = TaskUpdate(
status=TaskStatus.DONE,
completed=True
)
task_in = TaskUpdate(status=TaskStatus.DONE, completed=True)
return self.update(db, db_obj=task, obj_in=task_in)
task = CRUDTask(Task)
task = CRUDTask(Task)

74
app/crud/user.py Normal file
View File

@ -0,0 +1,74 @@
from typing import Any, Dict, Optional, Union
from sqlalchemy import or_
from sqlalchemy.orm import Session
from app.core.security import get_password_hash, verify_password
from app.crud.base import CRUDBase
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
def get_by_email(self, db: Session, *, email: str) -> Optional[User]:
return db.query(User).filter(User.email == email).first()
def get_by_username(self, db: Session, *, username: str) -> Optional[User]:
return db.query(User).filter(User.username == username).first()
def get_by_email_or_username(
self, db: Session, *, email_or_username: str
) -> Optional[User]:
return (
db.query(User)
.filter(
or_(User.email == email_or_username, User.username == email_or_username)
)
.first()
)
def create(self, db: Session, *, obj_in: UserCreate) -> User:
db_obj = User(
email=obj_in.email,
username=obj_in.username,
hashed_password=get_password_hash(obj_in.password),
is_active=obj_in.is_active,
is_superuser=obj_in.is_superuser,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
self, db: Session, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
) -> User:
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.model_dump(exclude_unset=True)
if "password" in update_data and update_data["password"]:
hashed_password = get_password_hash(update_data["password"])
del update_data["password"]
update_data["hashed_password"] = hashed_password
return super().update(db, db_obj=db_obj, obj_in=update_data)
def authenticate(
self, db: Session, *, email_or_username: str, password: str
) -> Optional[User]:
user = self.get_by_email_or_username(db, email_or_username=email_or_username)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
def is_active(self, user: User) -> bool:
return user.is_active
def is_superuser(self, user: User) -> bool:
return user.is_superuser
# Create instance for use throughout the app
user = CRUDUser(User)

3
app/db/base.py
View File

@ -1,4 +1,5 @@
# Import all the models, so that Base has them before being
# imported by Alembic
from app.db.base_class import Base # noqa
from app.models.task import Task # noqa
from app.models.task import Task # noqa
from app.models.user import User # noqa

4
app/db/base_class.py
View File

@ -7,8 +7,8 @@ from sqlalchemy.ext.declarative import as_declarative, declared_attr
class Base:
id: Any
__name__: str
# Generate __tablename__ automatically based on class name
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()
return cls.__name__.lower()

258
app/db/init_db.py
View File

@ -1,12 +1,8 @@
import os
import sys
import time
import sqlite3
from pathlib import Path
from datetime import datetime, timedelta
from sqlalchemy import inspect, text
from sqlalchemy.exc import OperationalError
from sqlalchemy import text
from app.db.base import Base # Import all models
from app.db.session import engine, db_file
@ -21,22 +17,22 @@ def init_db() -> None:
print(f"Initializing database at {db_file}")
print(f"Using SQLAlchemy URL: {settings.SQLALCHEMY_DATABASE_URL}")
print(f"DB_DIR is set to: {DB_DIR} (this should be /app/db in production)")
# First try direct SQLite approach to ensure we have a basic database file
try:
# Ensure database file exists and is writable
with open(db_file, 'a'): # Try opening for append (creates if doesn't exist)
with open(db_file, "a"): # Try opening for append (creates if doesn't exist)
os.utime(db_file, None) # Update access/modify time
print(f"Database file exists and is writable: {db_file}")
# Try direct SQLite connection to create task table
conn = sqlite3.connect(str(db_file))
# Enable foreign keys and WAL journal mode
conn.execute("PRAGMA foreign_keys = ON")
conn.execute("PRAGMA journal_mode = WAL")
# Create task table if it doesn't exist
conn.execute("""
CREATE TABLE IF NOT EXISTS task (
@ -48,33 +44,89 @@ def init_db() -> None:
due_date TEXT,
completed INTEGER DEFAULT 0,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
user_id INTEGER
)
""")
# Create user table if it doesn't exist
conn.execute("""
CREATE TABLE IF NOT EXISTS user (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL UNIQUE,
username TEXT NOT NULL UNIQUE,
hashed_password TEXT NOT NULL,
is_active INTEGER DEFAULT 1,
is_superuser INTEGER DEFAULT 0,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Create an index on the id column
# Create indexes
conn.execute("CREATE INDEX IF NOT EXISTS idx_task_id ON task(id)")
# Add a sample task if the table is empty
conn.execute("CREATE INDEX IF NOT EXISTS idx_task_user_id ON task(user_id)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_user_id ON user(id)")
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_user_email ON user(email)")
conn.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_user_username ON user(username)"
)
# Add a default admin user if the user table is empty
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM task")
count = cursor.fetchone()[0]
if count == 0:
cursor.execute("SELECT COUNT(*) FROM user")
user_count = cursor.fetchone()[0]
if user_count == 0:
now = datetime.utcnow().isoformat()
conn.execute("""
INSERT INTO task (title, description, priority, status, completed, created_at, updated_at)
# Import get_password_hash function here to avoid F823 error
from app.core.security import get_password_hash
admin_password_hash = get_password_hash("adminpassword")
conn.execute(
"""
INSERT INTO user (email, username, hashed_password, is_active, is_superuser, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
"Example Task",
"This is an example task created during initialization",
"medium",
"todo",
0,
now,
now
))
""",
(
"admin@example.com",
"admin",
admin_password_hash,
1, # is_active
1, # is_superuser
now,
now,
),
)
print("Created default admin user: admin@example.com / adminpassword")
# Add a sample task if the task table is empty
cursor.execute("SELECT COUNT(*) FROM task")
task_count = cursor.fetchone()[0]
if task_count == 0:
# Get the admin user ID if it exists
cursor.execute("SELECT id FROM user WHERE username = ?", ("admin",))
admin_row = cursor.fetchone()
admin_id = admin_row[0] if admin_row else None
now = datetime.utcnow().isoformat()
conn.execute(
"""
INSERT INTO task (title, description, priority, status, completed, created_at, updated_at, user_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"Example Task",
"This is an example task created during initialization",
"medium",
"todo",
0,
now,
now,
admin_id,
),
)
conn.commit()
cursor.close()
conn.close()
@ -82,58 +134,99 @@ def init_db() -> None:
except Exception as e:
print(f"Error during direct SQLite initialization: {e}")
import traceback
print(traceback.format_exc())
# Now try with SQLAlchemy as a backup approach
try:
print("Attempting SQLAlchemy database initialization...")
# Try to create all tables from models
Base.metadata.create_all(bind=engine)
print("Successfully created tables with SQLAlchemy")
# Verify tables exist
with engine.connect() as conn:
# Get list of tables
result = conn.execute(text(
"SELECT name FROM sqlite_master WHERE type='table'"
))
result = conn.execute(
text("SELECT name FROM sqlite_master WHERE type='table'")
)
tables = [row[0] for row in result]
print(f"Tables in database: {', '.join(tables)}")
# Verify task table exists
if 'task' in tables:
# Check if task table is empty
result = conn.execute(text("SELECT COUNT(*) FROM task"))
task_count = result.scalar()
print(f"Task table contains {task_count} records")
# If table exists but is empty, add a sample task
if task_count == 0:
print("Adding sample task with SQLAlchemy")
from app.models.task import Task, TaskPriority, TaskStatus
sample_task = Task(
title="Sample SQLAlchemy Task",
description="This is a sample task created with SQLAlchemy",
priority=TaskPriority.MEDIUM,
status=TaskStatus.TODO,
completed=False,
# Verify user table exists
if "user" in tables:
# Check if user table is empty
result = conn.execute(text("SELECT COUNT(*) FROM user"))
user_count = result.scalar()
print(f"User table contains {user_count} records")
# If user table is empty, add default admin user
if user_count == 0:
print("Adding default admin user with SQLAlchemy")
from app.models.user import User
from app.core.security import get_password_hash
admin_user = User(
email="admin@example.com",
username="admin",
hashed_password=get_password_hash("adminpassword"),
is_active=True,
is_superuser=True,
created_at=datetime.utcnow(),
updated_at=datetime.utcnow()
updated_at=datetime.utcnow(),
)
from app.db.session import SessionLocal
db = SessionLocal()
db.add(sample_task)
db.add(admin_user)
db.commit()
# Get the admin user ID for the sample task
admin_id = None
admin = db.query(User).filter_by(username="admin").first()
if admin:
admin_id = admin.id
# Verify task table exists
if "task" in tables:
# Check if task table is empty
result = conn.execute(text("SELECT COUNT(*) FROM task"))
task_count = result.scalar()
print(f"Task table contains {task_count} records")
# If table exists but is empty, add a sample task
if task_count == 0 and admin_id:
print("Adding sample task with SQLAlchemy")
from app.models.task import Task, TaskPriority, TaskStatus
sample_task = Task(
title="Sample SQLAlchemy Task",
description="This is a sample task created with SQLAlchemy",
priority=TaskPriority.MEDIUM,
status=TaskStatus.TODO,
completed=False,
user_id=admin_id,
created_at=datetime.utcnow(),
updated_at=datetime.utcnow(),
)
db.add(sample_task)
db.commit()
print("Added sample task with SQLAlchemy")
db.close()
print("Added sample task with SQLAlchemy")
print("Added default admin user with SQLAlchemy")
else:
print("WARNING: 'user' table not found!")
if "task" not in tables:
print("WARNING: 'task' table not found!")
print("SQLAlchemy database initialization completed")
except Exception as e:
print(f"Error during SQLAlchemy initialization: {e}")
import traceback
print(traceback.format_exc())
print("Continuing despite SQLAlchemy initialization error...")
@ -146,17 +239,19 @@ def create_test_task():
try:
conn = sqlite3.connect(str(db_file))
cursor = conn.cursor()
# Check if task table exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='task'")
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='task'"
)
if not cursor.fetchone():
print("Task table doesn't exist - cannot create test task")
return
# Check if any tasks exist
cursor.execute("SELECT COUNT(*) FROM task")
count = cursor.fetchone()[0]
if count == 0:
# Create a task directly with SQLite
now = datetime.utcnow().isoformat()
@ -166,47 +261,51 @@ def create_test_task():
title, description, priority, status,
completed, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?)
""",
""",
(
"Test Task (Direct SQL)",
"Test Task (Direct SQL)",
"This is a test task created directly with SQLite",
"medium",
"todo",
0, # not completed
now,
now
)
now,
),
)
conn.commit()
task_id = cursor.lastrowid
print(f"Created test task with direct SQLite, ID: {task_id}")
else:
print(f"Found {count} existing tasks, no need to create test task")
conn.close()
except Exception as e:
print(f"Error with direct SQLite test task creation: {e}")
# Continue with SQLAlchemy approach
# Now try with SQLAlchemy
try:
from app.crud.task import task as task_crud
from app.schemas.task import TaskCreate
from app.db.session import SessionLocal
db = SessionLocal()
try:
# Check if there are any tasks
try:
existing_tasks = db.execute(text("SELECT COUNT(*) FROM task")).scalar()
existing_tasks = db.execute(
text("SELECT COUNT(*) FROM task")
).scalar()
if existing_tasks > 0:
print(f"Test task not needed, found {existing_tasks} existing tasks")
print(
f"Test task not needed, found {existing_tasks} existing tasks"
)
return
except Exception as e:
print(f"Error checking for existing tasks: {e}")
# Continue anyway to try creating a task
# Create a test task
test_task = TaskCreate(
title="Test Task (SQLAlchemy)",
@ -214,23 +313,24 @@ def create_test_task():
priority="medium",
status="todo",
due_date=datetime.utcnow() + timedelta(days=7),
completed=False
completed=False,
)
created_task = task_crud.create(db, obj_in=test_task)
print(f"Created test task with SQLAlchemy, ID: {created_task.id}")
finally:
db.close()
except Exception as e:
print(f"Error with SQLAlchemy test task creation: {e}")
except Exception as e:
print(f"Global error creating test task: {e}")
import traceback
print(traceback.format_exc())
if __name__ == "__main__":
init_db()
create_test_task()
create_test_task()

41
app/db/session.py
View File

@ -1,12 +1,9 @@
import os
import time
import sqlite3
from typing import Generator
from pathlib import Path
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from app.core.config import settings, DB_DIR
@ -22,19 +19,20 @@ try:
print(f"Database file created or verified: {db_file}")
except Exception as e:
print(f"Warning: Could not create database file: {e}")
# Configure SQLite connection with simplified, robust settings
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URL,
connect_args={
"check_same_thread": False,
"timeout": 30, # Wait up to 30 seconds for the lock
"timeout": 30, # Wait up to 30 seconds for the lock
},
# Minimal pool settings for stability
pool_pre_ping=True, # Verify connections before usage
echo=True, # Log all SQL for debugging
pool_pre_ping=True, # Verify connections before usage
echo=True, # Log all SQL for debugging
)
# Add essential SQLite optimizations
@event.listens_for(engine, "connect")
def optimize_sqlite_connection(dbapi_connection, connection_record):
@ -46,9 +44,11 @@ def optimize_sqlite_connection(dbapi_connection, connection_record):
except Exception as e:
print(f"Warning: Could not configure SQLite connection: {e}")
# Simplified Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# More robust database access with retry logic and error printing
def get_db() -> Generator:
"""
@ -56,34 +56,35 @@ def get_db() -> Generator:
"""
db = None
retries = 3
for attempt in range(retries):
try:
db = SessionLocal()
# Log connection attempt
print(f"Database connection attempt {attempt+1}")
print(f"Database connection attempt {attempt + 1}")
# Test connection with a simple query
db.execute("SELECT 1")
# Connection succeeded
print("Database connection successful")
yield db
break
except Exception as e:
# Close failed connection
if db:
db.close()
db = None
error_msg = f"Database connection attempt {attempt+1} failed: {e}"
error_msg = f"Database connection attempt {attempt + 1} failed: {e}"
print(error_msg)
# Log critical error details
import traceback
print(f"Error traceback: {traceback.format_exc()}")
# Check if we can directly access database
try:
# Try direct sqlite3 connection as a test
@ -93,19 +94,19 @@ def get_db() -> Generator:
print("Direct SQLite connection succeeded but SQLAlchemy failed")
except Exception as direct_e:
print(f"Direct SQLite connection also failed: {direct_e}")
# Last attempt - raise the error to return 500 status
if attempt == retries - 1:
print("All database connection attempts failed")
raise
# Otherwise sleep and retry
time.sleep(1)
# Always ensure db is closed
try:
if db:
print("Closing database connection")
db.close()
except Exception as e:
print(f"Error closing database: {e}")
print(f"Error closing database: {e}")

19
app/models/task.py
View File

@ -1,8 +1,17 @@
from datetime import datetime
from enum import Enum
from typing import Optional
from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean, Enum as SQLEnum
from sqlalchemy import (
Column,
Integer,
String,
Text,
DateTime,
Boolean,
Enum as SQLEnum,
ForeignKey,
)
from sqlalchemy.orm import relationship
from app.db.base_class import Base
@ -28,4 +37,8 @@ class Task(Base):
due_date = Column(DateTime, nullable=True)
completed = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Add relationship to User model
user_id = Column(Integer, ForeignKey("user.id"), nullable=True)
user = relationship("User", back_populates="tasks")

20
app/models/user.py Normal file
View File

@ -0,0 +1,20 @@
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Boolean
from sqlalchemy.orm import relationship
from app.db.base_class import Base
class User(Base):
id = Column(Integer, primary_key=True, index=True)
email = Column(String(255), unique=True, index=True, nullable=False)
username = Column(String(100), unique=True, index=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationship with Task model
tasks = relationship("Task", back_populates="user", cascade="all, delete-orphan")

14
app/schemas/task.py
View File

@ -13,7 +13,7 @@ class TaskBase(BaseModel):
status: TaskStatus = TaskStatus.TODO
due_date: Optional[datetime] = None
completed: bool = False
model_config = {
"json_encoders": {
datetime: lambda dt: dt.isoformat(),
@ -32,7 +32,7 @@ class TaskUpdate(BaseModel):
status: Optional[TaskStatus] = None
due_date: Optional[datetime] = None
completed: Optional[bool] = None
model_config = {
"json_encoders": {
datetime: lambda dt: dt.isoformat(),
@ -45,10 +45,10 @@ class TaskUpdate(BaseModel):
"description": "Updated task description",
"priority": "high",
"status": "in_progress",
"completed": False
"completed": False,
}
]
}
},
}
@ -57,10 +57,8 @@ class TaskInDBBase(TaskBase):
created_at: datetime
updated_at: datetime
model_config = {
"from_attributes": True
}
model_config = {"from_attributes": True}
class Task(TaskInDBBase):
pass
pass

61
app/schemas/user.py Normal file
View File

@ -0,0 +1,61 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, EmailStr, Field, validator
class UserBase(BaseModel):
email: Optional[EmailStr] = None
username: Optional[str] = None
is_active: Optional[bool] = True
is_superuser: Optional[bool] = False
class UserCreate(UserBase):
email: EmailStr = Field(..., description="User email")
username: str = Field(..., min_length=3, max_length=50, description="Username")
password: str = Field(..., min_length=8, description="Password")
@validator("username")
def username_valid(cls, v):
# Additional validation for username
if not v.isalnum():
raise ValueError("Username must be alphanumeric")
return v
class UserUpdate(UserBase):
password: Optional[str] = Field(None, min_length=8, description="Password")
class UserInDBBase(UserBase):
id: int
created_at: datetime
updated_at: datetime
model_config = {"from_attributes": True}
class User(UserInDBBase):
"""Return user information without sensitive data"""
pass
class UserInDB(UserInDBBase):
"""User model stored in DB, with hashed password"""
hashed_password: str
class Token(BaseModel):
"""OAuth2 compatible token"""
access_token: str
token_type: str
class TokenPayload(BaseModel):
"""JWT token payload"""
sub: Optional[int] = None

136
main.py
View File

@ -1,15 +1,14 @@
import sys
import os
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
# Add project root to Python path for imports in alembic migrations
project_root = Path(__file__).parent.absolute()
sys.path.insert(0, str(project_root))
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.api.routers import api_router
from app.core.config import settings
from app.db import init_db
@ -19,17 +18,19 @@ print("Starting database initialization...")
try:
# Get absolute path of the database file from your config
from app.db.session import db_file
print(f"Database path: {db_file}")
# Check directory permissions for the configured DB_DIR
from app.core.config import DB_DIR
print(f"Database directory: {DB_DIR}")
print(f"Database directory exists: {DB_DIR.exists()}")
print(f"Database directory is writable: {os.access(DB_DIR, os.W_OK)}")
# Initialize the database and create test task
init_db.init_db()
# Try to create a test task
try:
init_db.create_test_task()
@ -39,6 +40,7 @@ try:
except Exception as e:
print(f"Error initializing database: {e}")
import traceback
print(f"Detailed error: {traceback.format_exc()}")
# Continue with app startup even if DB init fails, to allow debugging
@ -53,33 +55,34 @@ app.add_middleware(
allow_headers=["*"],
)
# Add comprehensive exception handlers for better error reporting
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
import traceback
import sys
# Log the full error with traceback to stdout/stderr
error_tb = traceback.format_exc()
print(f"CRITICAL ERROR: {str(exc)}", file=sys.stderr)
print(f"Request path: {request.url.path}", file=sys.stderr)
print(f"Traceback:\n{error_tb}", file=sys.stderr)
# Get request info for debugging
headers = dict(request.headers)
# Remove sensitive headers
if 'authorization' in headers:
headers['authorization'] = '[REDACTED]'
if 'cookie' in headers:
headers['cookie'] = '[REDACTED]'
if "authorization" in headers:
headers["authorization"] = "[REDACTED]"
if "cookie" in headers:
headers["cookie"] = "[REDACTED]"
# Include minimal traceback in response for debugging
tb_lines = error_tb.split('\n')
tb_lines = error_tb.split("\n")
simplified_tb = []
for line in tb_lines:
if line and not line.startswith(' '):
if line and not line.startswith(" "):
simplified_tb.append(line)
# Create detailed error response
error_detail = {
"status": "error",
@ -87,29 +90,34 @@ async def global_exception_handler(request: Request, exc: Exception):
"type": str(type(exc).__name__),
"path": request.url.path,
"method": request.method,
"traceback_summary": simplified_tb[-10:] if len(simplified_tb) > 10 else simplified_tb,
"traceback_summary": simplified_tb[-10:]
if len(simplified_tb) > 10
else simplified_tb,
}
# Add SQLite diagnostic check
try:
import sqlite3
from app.db.session import db_file
# Try basic SQLite operations
conn = sqlite3.connect(str(db_file))
cursor = conn.cursor()
cursor.execute("PRAGMA integrity_check")
integrity = cursor.fetchone()[0]
# Check if task table exists
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='task'")
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='task'"
)
task_table_exists = cursor.fetchone() is not None
# Get file info
import os
file_exists = os.path.exists(db_file)
file_size = os.path.getsize(db_file) if file_exists else 0
# Add SQLite diagnostics to response
error_detail["db_diagnostics"] = {
"file_exists": file_exists,
@ -117,11 +125,11 @@ async def global_exception_handler(request: Request, exc: Exception):
"integrity": integrity,
"task_table_exists": task_table_exists,
}
conn.close()
except Exception as db_error:
error_detail["db_diagnostics"] = {"error": str(db_error)}
# Return the error response
print(f"Returning error response: {error_detail}")
return JSONResponse(
@ -129,6 +137,7 @@ async def global_exception_handler(request: Request, exc: Exception):
content=error_detail,
)
# Include the API router directly (no version prefix)
app.include_router(api_router)
@ -141,16 +150,23 @@ def api_info():
return {
"name": settings.PROJECT_NAME,
"version": "1.0.0",
"description": "A RESTful API for managing tasks",
"description": "A RESTful API for managing tasks with user authentication",
"endpoints": {
"authentication": {
"register": "/auth/register",
"login": "/auth/login",
"me": "/auth/me",
"test-token": "/auth/test-token",
},
"tasks": "/tasks",
"docs": "/docs",
"redoc": "/redoc",
"health": "/health",
"db_test": "/db-test"
}
"db_test": "/db-test",
},
}
@app.get("/health", tags=["health"])
def health_check():
"""
@ -172,54 +188,60 @@ def test_db_connection():
from sqlalchemy import text
from app.db.session import engine, db_file
from app.core.config import DB_DIR
# First check direct file access
file_info = {
"db_dir": str(DB_DIR),
"db_file": str(db_file),
"exists": os.path.exists(db_file),
"size": os.path.getsize(db_file) if os.path.exists(db_file) else 0,
"writable": os.access(db_file, os.W_OK) if os.path.exists(db_file) else False,
"dir_writable": os.access(DB_DIR, os.W_OK) if os.path.exists(DB_DIR) else False
"writable": os.access(db_file, os.W_OK)
if os.path.exists(db_file)
else False,
"dir_writable": os.access(DB_DIR, os.W_OK)
if os.path.exists(DB_DIR)
else False,
}
# Try direct SQLite connection
sqlite_test = {}
try:
conn = sqlite3.connect(str(db_file))
sqlite_test["connection"] = "successful"
# Check if task table exists
cursor = conn.cursor()
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = [row[0] for row in cursor.fetchall()]
sqlite_test["tables"] = tables
# Check for task table specifically
if 'task' in tables:
if "task" in tables:
cursor.execute("SELECT COUNT(*) FROM task")
task_count = cursor.fetchone()[0]
sqlite_test["task_count"] = task_count
# Get a sample task if available
if task_count > 0:
cursor.execute("SELECT * FROM task LIMIT 1")
column_names = [description[0] for description in cursor.description]
column_names = [
description[0] for description in cursor.description
]
row = cursor.fetchone()
sample_task = dict(zip(column_names, row))
sqlite_test["sample_task"] = sample_task
# Check database integrity
cursor.execute("PRAGMA integrity_check")
integrity = cursor.fetchone()[0]
sqlite_test["integrity"] = integrity
conn.close()
except Exception as e:
sqlite_test["connection"] = "failed"
sqlite_test["error"] = str(e)
sqlite_test["traceback"] = traceback.format_exc()
# Try SQLAlchemy connection
sqlalchemy_test = {}
try:
@ -227,32 +249,38 @@ def test_db_connection():
# Basic connectivity test
result = conn.execute(text("SELECT 1")).scalar()
sqlalchemy_test["basic_query"] = result
# Check tables
result = conn.execute(text("SELECT name FROM sqlite_master WHERE type='table'"))
result = conn.execute(
text("SELECT name FROM sqlite_master WHERE type='table'")
)
tables = [row[0] for row in result]
sqlalchemy_test["tables"] = tables
# Check task table
if 'task' in tables:
if "task" in tables:
result = conn.execute(text("SELECT COUNT(*) FROM task"))
sqlalchemy_test["task_count"] = result.scalar()
except Exception as e:
sqlalchemy_test["connection"] = "failed"
sqlalchemy_test["error"] = str(e)
sqlalchemy_test["traceback"] = traceback.format_exc()
# Check environment
env_info = {
"cwd": os.getcwd(),
"env_variables": {k: v for k, v in os.environ.items() if k.startswith(('DB_', 'SQL', 'PATH'))}
"env_variables": {
k: v
for k, v in os.environ.items()
if k.startswith(("DB_", "SQL", "PATH"))
},
}
# Try to create a test file
write_test = {}
try:
test_path = DB_DIR / "write_test.txt"
with open(test_path, 'w') as f:
with open(test_path, "w") as f:
f.write("Test content")
write_test["success"] = True
write_test["path"] = str(test_path)
@ -260,7 +288,7 @@ def test_db_connection():
except Exception as e:
write_test["success"] = False
write_test["error"] = str(e)
return {
"status": "ok",
"timestamp": datetime.datetime.utcnow().isoformat(),
@ -268,13 +296,13 @@ def test_db_connection():
"sqlite_test": sqlite_test,
"sqlalchemy_test": sqlalchemy_test,
"environment": env_info,
"write_test": write_test
"write_test": write_test,
}
except Exception as e:
# Catch-all error handler
return {
"status": "error",
"message": str(e),
"traceback": traceback.format_exc()
}
"traceback": traceback.format_exc(),
}

6
requirements.txt
View File

@ -5,4 +5,8 @@ alembic>=1.12.0
pydantic>=2.4.2
pydantic-settings>=2.0.3
python-multipart>=0.0.6
ruff>=0.1.3
ruff>=0.1.3
passlib>=1.7.4
bcrypt>=4.0.1
python-jose>=3.3.0
email-validator>=2.0.0