#подключение библиотек from fastapi import FastAPI, Response, Path, Depends, Body, Query, status, HTTPException, APIRouter from fastapi.responses import HTMLResponse, FileResponse, JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer from fastapi.templating import Jinja2Templates from fastapi.requests import Request from sqlalchemy import create_engine from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session from pydantic import json from starlette.staticfiles import StaticFiles from app.utils import * import requests from uuid import uuid4 SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db" engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} ) Base = declarative_base() class UserDatabase(Base):#класс пользователя __tablename__ = "users" id = Column(Integer, primary_key=True, index=True)#айди пользователя phone = Column(Integer, nullable=True)#номер телефона пользователя email = Column(String)#электронная почта пользователя password = Column(String) # пароль hashed_password = Column(String) name = Column(String, nullable=True)#имя пользователя surname = Column(String)#фамилия пользователя class Announcement(Base): #класс объявления __tablename__ = "announcements" id = Column(Integer, primary_key=True, index=True)#айди объявления user_id = Column(Integer)#айди создателя объявления name = Column(String) # название объявления category = Column(String)#категория продукта из объявления best_by = Column(Integer)#срок годности продукта из объявления adress = Column(String) longtitude = Column(Integer) latitude = Column(Integer) description = Column(String)#описание продукта в объявлении src = Column(String, nullable=True) #изображение продукта в объявлении metro = Column(String)#ближайщее метро от адреса нахождения продукта trashId = Column(Integer, nullable=True) booked_by = Column(Integer)#статус бронирования (либо -1, либо айди бронирующего) class Trashbox(Base):#класс мусорных баков __tablename__ = "trashboxes" id = Column(Integer, primary_key=True, index=True)#айди name = Column(String, nullable=True)#имя пользователя adress = Column(String) latitude = Column(Integer) longtitude = Column(Integer) category = Column(String)#категория продукта из объявления Base.metadata.create_all(bind=engine) SessionLocal = sessionmaker(autoflush=True, bind=engine) db = SessionLocal() # Пробный чувак tom = UserDatabase(name="Tom", phone="89999999", email="pupka", password="1234", surname="Smith") # db.add(tom) # добавляем в бд # db.commit() # сохраняем изменения # db.refresh(tom) # обновляем состояние объекта # Пробное объявление 1 a1 = Announcement(user_id=1, category="cat", best_by="201223", adress="abd", longtitude=23, latitude=22, description="abv", src="111", metro="Lesnaya", booked_by=2) # Пробное объявление 2 a2 = Announcement(user_id=1, category="dog", best_by="221223", adress="abd", longtitude=50, latitude=12, description="vvv", src="110", metro="Petrogradskaya", booked_by=2) a3 = Announcement(user_id=1, category="a", best_by="221223", adress="abd", longtitude=20, latitude=25, description="vvv", src="101", metro="metro", booked_by=2) trash1 = Trashbox(name="Tom", adress="abd", longtitude=23, latitude=22, category="indisposable") # db.add(a1) # добавляем в бд # db.add(a2) # добавляем в бд # db.add(a3) # добавляем в бд # db.add(trash1) # добавляем в бд # db.commit() # сохраняем изменения # db.refresh(a1) # обновляем состояние объекта # db.refresh(a2) # обновляем состояние объекта # db.refresh(a3) # обновляем состояние объекта # db.refresh(trash1) # обновляем состояние объекта # # Удалить все # db.query(User).delete() # db.query(Announcement).delete() # db.commit() # Непосредственно преложение app = FastAPI() templates = Jinja2Templates(directory="./front/dist") app.mount("/static", StaticFiles(directory = "./front/dist")) @app.get("/api/announcements")#адрес объявлений def annoncements_list(user_id: int = None, metro: str = None, category: str = None, booked_by: int = -1): # Считываем данные из Body и отображаем их на странице. # В последствии будем вставлять данные в html-форму a = db.query(Announcement) b = db.query(Announcement) c = db.query(Announcement) d = db.query(Announcement) e = db.query(Announcement) if user_id != None: b = a.filter(Announcement.user_id == user_id) if metro != None: c = a.filter(Announcement.metro == metro) if category != None: d = a.filter(Announcement.category == category) if booked_by != -1: e = a.filter(Announcement.booked_by == booked_by) if not any([category, user_id, metro]) and booked_by == -1: result = a.all() else: result = b.intersect(c, d, e).all() return {"Success" : True, "list_of_announcements": result} @app.get("/api/announcement")#адрес объявлений def single_annoncement(user_id:int): # Считываем данные из Body и отображаем их на странице. # В последствии будем вставлять данные в html-форму try: annoncement = db.get(Announcement, user_id) return {"id": annoncement.id, "user_id": annoncement.user_id, "name": annoncement.name, "category": annoncement.category, "best_by": annoncement.best_by, "adress": annoncement.adress, "description": annoncement.description, "metro": annoncement.metro, "latitude": annoncement.latitude, "longtitude":annoncement.longtitude, "trashId": annoncement.trashId, "src":annoncement.src, "booked_by":annoncement.booked_by} except: return {"Answer" : False} #если неуданый доступ, то сообщаем об этом # Занести объявление в базу @app.put("/api/announcement")#адрес объявлений def put_in_db(data = Body()): try: temp_ancmt = Announcement(data.id, data.userId, data.name, data.category, data.bestBy, data.adress, data.longtitude, data.latitude, data.description, data.src, data.metro, data.trashId, data.bookedBy) db.add(temp_ancmt) # добавляем в бд db.commit() # сохраняем изменения db.refresh(temp_ancmt) # обновляем состояние объекта return {"Answer" : True} except: return {"Answer" : False} # Удалить объявления из базы @app.delete("/api/announcement") #адрес объявления def delete_from_db(data = Body()):#функция удаления объекта из БД try: db.delete(user_id=data.user_id)#удаление из БД db.commit() # сохраняем изменения return {"Answer" : True} except: return {"Answer" : False} # Забронировать объявление @app.put("/api/book") def change_book_status(data = Body()): # Получаем id пользователя, который бронирует объявление temp_user_id = 1 # Находим объявление по данному id announcement_to_change = db.query(Announcement).filter(user_id == temp_user_id).first() # Изменяем поле booked_status на полученный id announcement_to_change.booked_status = temp_user_id return {"Success": True} @app.post("/api/signup") def create_user(data = Body()): if db.query(UserDatabase).filter(User.email == data["email"]).first() == None: new_user = UserDatabase(id=data["id"], email=data["email"], password=data["password"], name=data["name"], surname=data["surname"]) db.add(new_user) db.commit() db.refresh(new_user) # обновляем состояние объекта return {"Success": True} return {"Success": False, "Message": "Пользователь с таким email уже зарегестрирован."} @app.post("/api/token", response_model=Token) async def login_for_access_token( form_data: Annotated[OAuth2PasswordRequestForm, Depends()] ): user = authenticate_user(db.query(UserDatabase).all(), form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"user_id": user.id}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.get("/api/users/me/", response_model=User) async def read_users_me( current_user: Annotated[User, Depends(get_current_active_user)] ): return current_user @app.get("/api/users/me/items/") async def read_own_items( current_user: Annotated[User, Depends(get_current_active_user)] ): return [{"Current user name": current_user.name, "Current user surname": current_user.surname}] @app.get("/api/trashbox") def get_trashboxes(lat:float, lng:float):#крутая функция для работы с api BASE_URL='https://geointelect2.gate.petersburg.ru'#адрес сайта и мой токин my_token='eyJhbGciOiJSUzI1NiIsInR5cCIgOiAiSldUIiwia2lkIiA6ICJhU1RaZm42bHpTdURYcUttRkg1SzN5UDFhT0FxUkhTNm9OendMUExaTXhFIn0.eyJleHAiOjE3Nzg2NTk4MjEsImlhdCI6MTY4Mzk2NTQyMSwianRpIjoiOTI2ZGMyNmEtMGYyZi00OTZiLWI0NTUtMWQyYWM5YmRlMTZkIiwiaXNzIjoiaHR0cHM6Ly9rYy5wZXRlcnNidXJnLnJ1L3JlYWxtcy9lZ3MtYXBpIiwiYXVkIjoiYWNjb3VudCIsInN1YiI6ImJjYjQ2NzljLTU3ZGItNDU5ZC1iNWUxLWRlOGI4Yzg5MTMwMyIsInR5cCI6IkJlYXJlciIsImF6cCI6ImFkbWluLXJlc3QtY2xpZW50Iiwic2Vzc2lvbl9zdGF0ZSI6IjE2MGU1ZGVkLWFmMjMtNDkyNS05OTc1LTRhMzM0ZjVmNTkyOSIsImFjciI6IjEiLCJhbGxvd2VkLW9yaWdpbnMiOlsiLyoiXSwicmVhbG1fYWNjZXNzIjp7InJvbGVzIjpbImRlZmF1bHQtcm9sZXMtZWdzLWFwaSIsIm9mZmxpbmVfYWNjZXNzIiwidW1hX2F1dGhvcml6YXRpb24iXX0sInJlc291cmNlX2FjY2VzcyI6eyJhY2NvdW50Ijp7InJvbGVzIjpbIm1hbmFnZS1hY2NvdW50IiwibWFuYWdlLWFjY291bnQtbGlua3MiLCJ2aWV3LXByb2ZpbGUiXX19LCJzY29wZSI6ImVtYWlsIHByb2ZpbGUiLCJzaWQiOiIxNjBlNWRlZC1hZjIzLTQ5MjUtOTk3NS00YTMzNGY1ZjU5MjkiLCJlbWFpbF92ZXJpZmllZCI6ZmFsc2UsIm5hbWUiOiLQktC70LDQtNC40LzQuNGAINCv0LrQvtCy0LvQtdCyIiwicHJlZmVycmVkX3VzZXJuYW1lIjoiZTBmYzc2OGRhOTA4MjNiODgwZGQzOGVhMDJjMmQ5NTciLCJnaXZlbl9uYW1lIjoi0JLQu9Cw0LTQuNC80LjRgCIsImZhbWlseV9uYW1lIjoi0K_QutC-0LLQu9C10LIifQ.BRyUIyY-KKnZ9xqTNa9vIsfKF0UN2VoA9h4NN4y7IgBVLiiS-j43QbeE6qgjIQo0pV3J8jtCAIPvJbO-Ex-GNkw_flgMiGHhKEpsHPW3WK-YZ-XsZJzVQ_pOmLte-Kql4z97WJvolqiXT0nMo2dlX2BGvNs6JNbupvcuGwL4YYpekYAaFNYMQrxi8bSN-R7FIqxP-gzZDAuQSWRRSUqVBLvmgRhphTM-FAx1sX833oXL9tR7ze3eDR_obSV0y6cKVIr4eIlKxFd82qiMrN6A6CTUFDeFjeAGERqeBPnJVXU36MHu7Ut7eOVav9OUARARWRkrZRkqzTfZ1iqEBq5Tsg' head = {'Authorization': 'Bearer {}'.format(my_token)} my_data={ 'x' : f"{lat}", 'y' : f"{lng}", 'limit' : '1' } response = requests.post(f"{BASE_URL}/nearest_recycling/get", headers=head, data=my_data) infos = response.json() trashboxes = [] for trashbox in infos["results"]: temp_dict = {} temp_dict["Category"] = trashbox["Category"] for obj in trashbox["Objects"]: coord_list = obj["geometry"] temp_dict["Lat"] = coord_list["coordinates"][0] temp_dict["Lng"] = coord_list["coordinates"][1] properties = obj["properties"] temp_dict["Name"] = properties["title"] temp_dict["Adress"] = properties["address"] trashboxes.append(temp_dict) return JSONResponse(trashboxes) @app.get("/{rest_of_path:path}") async def react_app(req: Request, rest_of_path: str): return templates.TemplateResponse('index.html', { 'request': req })