Coverage for src \ main.py: 94%

1from fastapi import FastAPI, Depends, HTTPException 

2from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm 

3from sqlalchemy import create_engine, Column, Integer, String, DateTime 

4from sqlalchemy.ext.declarative import declarative_base 

5from sqlalchemy.orm import sessionmaker, Session 

6from datetime import datetime, timedelta 

7from jose import jwt 

8from passlib.context import CryptContext 

9 

10# db 

11SQLALCHEMY_DATABASE_URL = 'sqlite:///./tasks.db' 

12engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False}) 

13SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) 

14Base = declarative_base() 

15 

16class User(Base): 

17 __tablename__ = 'users' 

18 id = Column(Integer, primary_key=True, index=True) 

19 username = Column(String, unique=True, index=True) 

20 hashed_password = Column(String) 

21 

22class Task(Base): 

23 __tablename__ = 'tasks' 

24 id = Column(Integer, primary_key=True, index=True) 

25 title = Column(String) 

26 description = Column(String, nullable=True) 

27 status = Column(String, default='pending') 

28 priority = Column(String, default='medium') 

29 created_at = Column(DateTime, default=datetime.now) 

30 owner_id = Column(Integer) 

31 

32Base.metadata.create_all(bind=engine) 

33 

34#authentification 

35SECRET_KEY = 'secret' 

36ALGORITHM = 'HS256' 

37pwd_context = CryptContext(schemes=['pbkdf2_sha256'], deprecated='auto') 

38oauth2_scheme = OAuth2PasswordBearer(tokenUrl='token') 

39 

40def get_db(): 

41 db = SessionLocal() 

42 try: 

43 yield db 

44 finally: 

45 db.close() 

46 

47def get_password_hash(password): 

48 return pwd_context.hash(password) 

49 

50def verify_password(plain_password, hashed_password): 

51 return pwd_context.verify(plain_password, hashed_password) 

52 

53def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)): 

54 try: 

55 username = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]).get('sub') 

56 if username is None: 

57 raise HTTPException(status_code=401, detail='Invalid token') 

58 user = db.query(User).filter(User.username == username).first() 

59 if user is None: 

60 raise HTTPException(status_code=401, detail='User not found') 

61 return user 

62 except: 

63 raise HTTPException(status_code=401, detail='Invalid token') 

64 

65#fastapi 

66app = FastAPI() 

67 

68@app.post('/register') 

69def register(username: str, password: str, db: Session = Depends(get_db)): 

70 existing_user = db.query(User).filter(User.username == username).first() 

71 if existing_user: 

72 raise HTTPException(status_code=400, detail='Username already exists') 

73 

74 new_user = User(username=username, hashed_password=get_password_hash(password)) 

75 db.add(new_user) 

76 db.commit() 

77 

78 return {'id': new_user.id, 'username': new_user.username} 

79 

80@app.post('/token') 

81def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)): 

82 user = db.query(User).filter(User.username == form_data.username).first() 

83 if not user or not verify_password(form_data.password, user.hashed_password): 

84 raise HTTPException(status_code=401, detail='Incorrect username or password') 

85 

86 token_data = {'sub': user.username, 'exp': datetime.now() + timedelta(minutes=30)} 

87 token = jwt.encode(token_data, SECRET_KEY, algorithm=ALGORITHM) 

88 

89 return {'access_token': token, 'token_type': 'bearer'} 

90 

91@app.post('/tasks') 

92def create_task(title: str, description: str = None, priority: str = 'medium', db: Session = Depends(get_db), 

93 current_user=Depends(get_current_user)): 

94 new_task = Task(title=title, description=description, priority=priority, owner_id=current_user.id) 

95 db.add(new_task) 

96 db.commit() 

97 db.refresh(new_task) 

98 

99 return { 

100 'id': new_task.id, 

101 'title': new_task.title, 

102 'description': new_task.description, 

103 'status': new_task.status, 

104 'priority': new_task.priority, 

105 'created_at': new_task.created_at, 

106 'owner_id': new_task.owner_id 

107 } 

108 

109@app.get('/tasks') 

110def get_tasks(sort_by: str = None, sort_desc: bool = False, search: str = None, db: Session = Depends(get_db), 

111 current_user=Depends(get_current_user)): 

112 tasks_query = db.query(Task).filter(Task.owner_id == current_user.id) 

113 

114 if search: 

115 tasks_query = tasks_query.filter(Task.title.contains(search) | Task.description.contains(search)) 

116 

117 if sort_by: 

118 if sort_by == 'title': 

119 if sort_desc: 

120 tasks_query = tasks_query.order_by(Task.title.desc()) 

121 else: 

122 tasks_query = tasks_query.order_by(Task.title) 

123 elif sort_by == 'status': 

124 if sort_desc: 

125 tasks_query = tasks_query.order_by(Task.status.desc()) 

126 else: 

127 tasks_query = tasks_query.order_by(Task.status) 

128 elif sort_by == 'created_at': 

129 if sort_desc: 

130 tasks_query = tasks_query.order_by(Task.created_at.desc()) 

131 else: 

132 tasks_query = tasks_query.order_by(Task.created_at) 

133 

134 tasks = tasks_query.all() 

135 return [ 

136 { 

137 'id': task.id, 

138 'title': task.title, 

139 'description': task.description, 

140 'status': task.status, 

141 'priority': task.priority, 

142 'created_at': task.created_at, 

143 'owner_id': task.owner_id 

144 } 

145 for task in tasks 

146 ] 

147 

148@app.get('/tasks/top/{n}') 

149def get_top_tasks(n: int, db: Session = Depends(get_db), current_user=Depends(get_current_user)): 

150 all_tasks = db.query(Task).filter(Task.owner_id == current_user.id).all() 

151 priority_values = {'high': 1, 'medium': 2, 'low': 3} 

152 all_tasks.sort(key=lambda task: priority_values.get(task.priority, float('inf'))) 

153 top_tasks = all_tasks[:n] 

154 return [ 

155 { 

156 'id': task.id, 

157 'title': task.title, 

158 'description': task.description, 

159 'status': task.status, 

160 'priority': task.priority, 

161 'created_at': task.created_at, 

162 'owner_id': task.owner_id 

163 } 

164 for task in top_tasks 

165 ] 

166 

167@app.get('/tasks/{task_id}') 

168def get_task(task_id: int, db: Session = Depends(get_db), current_user=Depends(get_current_user)): 

169 task = db.query(Task).filter(Task.id == task_id, Task.owner_id == current_user.id).first() 

170 

171 if not task: 

172 raise HTTPException(status_code=404, detail='Task not found') 

173 

174 return { 

175 'id': task.id, 

176 'title': task.title, 

177 'description': task.description, 

178 'status': task.status, 

179 'priority': task.priority, 

180 'created_at': task.created_at, 

181 'owner_id': task.owner_id 

182 } 

183 

184@app.put('/tasks/{task_id}') 

185def update_task(task_id: int, title: str = None, description: str = None, status: str = None, 

186 priority: str = None, db: Session = Depends(get_db), current_user=Depends(get_current_user)): 

187 task = db.query(Task).filter(Task.id == task_id, Task.owner_id == current_user.id).first() 

188 

189 if not task: 

190 raise HTTPException(status_code=404, detail='Task not found') 

191 

192 if title is not None: 

193 task.title = title 

194 if description is not None: 

195 task.description = description 

196 if status is not None: 

197 task.status = status 

198 if priority is not None: 

199 task.priority = priority 

200 

201 db.commit() 

202 db.refresh(task) 

203 

204 return { 

205 'id': task.id, 

206 'title': task.title, 

207 'description': task.description, 

208 'status': task.status, 

209 'priority': task.priority, 

210 'created_at': task.created_at, 

211 'owner_id': task.owner_id 

212 } 

213 

214@app.delete('/tasks/{task_id}') 

215def delete_task(task_id: int, db: Session = Depends(get_db), current_user=Depends(get_current_user)): 

216 task = db.query(Task).filter(Task.id == task_id, Task.owner_id == current_user.id).first() 

217 

218 if not task: 

219 raise HTTPException(status_code=404, detail='Task not found') 

220 

221 db.delete(task) 

222 db.commit() 

223 

224 return {'message': 'Task deleted successfully'}