created Todo CRUD and also apply middleware #7
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
…tegration with alembic
AbdullahCrypthon
suggested changes
Oct 22, 2025
Comment on lines
+24
to
+25
| op.add_column('todos', sa.Column('user_id', sa.Integer(), nullable=True)) | ||
| op.create_foreign_key(None, 'todos', 'user', ['user_id'], ['id']) |
There was a problem hiding this comment.
why you named todos for user relation here
| def downgrade() -> None: | ||
| """Downgrade schema.""" | ||
| # ### commands auto generated by Alembic - please adjust! ### | ||
| op.drop_constraint(None, 'todos', type_='foreignkey') |
There was a problem hiding this comment.
instead of having constraint name none have some name
| def upgrade() -> None: | ||
| """Upgrade schema.""" | ||
| # ### commands auto generated by Alembic - please adjust! ### | ||
| op.add_column('company', sa.Column('company_type', sa.String(), nullable=True)) |
| def upgrade() -> None: | ||
| """Upgrade schema.""" | ||
| # ### commands auto generated by Alembic - please adjust! ### | ||
| op.create_table('todos', |
app/auth.py
Outdated
|
|
||
| SECRET_KEY = "af3287c8391bb9f4f7a72feb3b85f72e1d5bd07cbf4fa4ad9497c78412923312" | ||
| ALGORITHM = "HS256" | ||
| ACCESS_TOKEN_EXPIRE_MINUTES = 7 * 24 * 60 |
There was a problem hiding this comment.
correct the format as i have defined you before not multiplications just value
app/services/company_service.py
Outdated
| company.location = location | ||
|
|
||
| self.repo.update(company) | ||
| return {"message": "Company updated successfully"} |
There was a problem hiding this comment.
same as above dont need hard coded reponse
app/services/company_service.py
Outdated
Comment on lines
40
to
44
| if not company: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Company not found") | ||
|
|
||
| self.repo.delete(company) | ||
| return {"message": "Company deleted successfully"} |
app/services/product_service.py
Outdated
Comment on lines
10
to
52
| def create_product(self, company_id: int, name: str, price: float, description: str): | ||
| existing_product = self.repo.get_by_name_and_company(company_id, name) | ||
| if existing_product: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_400_BAD_REQUEST, | ||
| detail="This product already exists for the company" | ||
| ) | ||
|
|
||
| product = Product(company_id=company_id, name=name, price=price, description=description) | ||
| created_product = self.repo.create(product) | ||
| return created_product | ||
|
|
||
| def list_products(self, skip: int, limit: int): | ||
| products = self.repo.get_all(skip, limit) | ||
| if not products: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No products found") | ||
| return products | ||
|
|
||
| def get_product(self, product_id: int): | ||
| product = self.repo.get_by_id(product_id) | ||
| if not product: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found") | ||
| return product | ||
|
|
||
| def update_product(self, product_id: int, name: str, price: float, description: str): | ||
| product = self.repo.get_by_id(product_id) | ||
| if not product: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found") | ||
|
|
||
| product.name = name | ||
| product.price = price | ||
| product.description = description | ||
|
|
||
| updated = self.repo.update(product) | ||
| return {"message": "Product updated successfully", "data": updated} | ||
|
|
||
| def delete_product(self, product_id: int): | ||
| product = self.repo.get_by_id(product_id) | ||
| if not product: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Product not found") | ||
|
|
||
| self.repo.delete(product) | ||
| return {"message": "Product deleted successfully"} |
app/services/todo_service.py
Outdated
Comment on lines
9
to
43
| def get_all_todo(self,user_id:int,skip: int, limit: int): | ||
| todos = self.repo.list_todo(user_id,skip, limit) | ||
| if not todos: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="No todos found") | ||
|
|
||
| return todos | ||
|
|
||
| def create_todo(self, title: str, description: str, published: bool, user_id: int): | ||
| existing = self.repo.get_todo_by_title(title) | ||
| if existing: | ||
| raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Todo already exists") | ||
|
|
||
| todo = Todo(title=title, description=description, published=published, user_id=user_id) | ||
| create_todo = self.repo.create(todo) | ||
| return create_todo | ||
|
|
||
| def update_todo(self, todo_id: int, title: str, description: str, published: bool): | ||
| todo = self.repo.get_todo_by_id(todo_id) | ||
| if not todo: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found") | ||
|
|
||
| todo.title = title | ||
| todo.description = description | ||
| todo.published = published | ||
|
|
||
| updated = self.repo.update(todo) | ||
| return {"message": "Todo updated successfully", "data": updated} | ||
|
|
||
| def delete_todo(self, todo_id: int): | ||
| todo = self.repo.get_todo_by_id(todo_id) | ||
| if not todo: | ||
| raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found") | ||
|
|
||
| self.repo.delete(todo) | ||
| return {"message": "Todo deleted successfully"} |
There was a problem hiding this comment.
correct these aswell accordingly
app/services/user_service.py
Outdated
Comment on lines
6
to
39
| class UserService: | ||
| def __init__(self, db: Session): | ||
| self.repo = UserRepository(db) | ||
|
|
||
| def create_user(self, username: str, hashed_pw: str): | ||
| existing_user = self.repo.get_by_username(username) | ||
| if existing_user: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_400_BAD_REQUEST, | ||
| detail="Username already exists" | ||
| ) | ||
|
|
||
| self.repo.create(username, hashed_pw) | ||
| return {"message": "User created successfully"} | ||
|
|
||
| def get_user(self, user_id: int): | ||
| user = self.repo.get_by_id(user_id) | ||
| if not user: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_404_NOT_FOUND, | ||
| detail="User not found" | ||
| ) | ||
| return user | ||
|
|
||
| def delete_user(self, user_id: int): | ||
| user = self.repo.get_by_id(user_id) | ||
| if not user: | ||
| raise HTTPException( | ||
| status_code=status.HTTP_404_NOT_FOUND, | ||
| detail="User not found" | ||
| ) | ||
|
|
||
| self.repo.delete(user) | ||
| return {"message": "User deleted successfully"} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
implement CRUD on todo list and also add middleware and authentuication