1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129
| from sqlalchemy import create_engine, Column, Integer, String, DateTime, JSON, UniqueConstraint, ForeignKey from sqlalchemy.orm import sessionmaker,declarative_base, relationship from datetime import datetime from sqlalchemy.orm import Session from sqlalchemy import select,text import pytz
DATABASE_URL="postgresql://uizbhnmkOOOOOO:K0r18XtU1Nb2yGwC1SfwoOOOOOOO@bngbuvea6wkwlOOOg-postgresql.services.clever-cloud.com:50OOO/bngbuvea6wOOOOOOhzig"
engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class Photo(Base): __tablename__ = 'photos_backup' id = Column(Integer, primary_key=True, index=True) user_id = Column(String(255)) group_id = Column(String(255)) photo_name = Column(String(255)) photo_url = Column(String(255)) photo_thumbnail_url = Column(String(255)) place_tmp= Column(String(50)) created_time = Column(DateTime, default=lambda: datetime.now(pytz.timezone('Asia/Taipei'))) photo_properties = Column(JSON) photo_image_hash = Column(String(255))
Base.metadata.create_all(bind=engine)
def add_photo(user_id, group_id, photo_name, photo_url, photo_thumbnail_url, place_tmp, photo_properties, photo_image_hash): db: Session = SessionLocal() try: new_photo = Photo( user_id=user_id, group_id=group_id, photo_name=photo_name, photo_url=photo_url, photo_thumbnail_url=photo_thumbnail_url, place_tmp=place_tmp, photo_properties=photo_properties, photo_image_hash=photo_image_hash ) db.add(new_photo) db.commit() db.refresh(new_photo) print(f"新增照片資料成功,ID: {new_photo.id}") except Exception as e: db.rollback() print(f"新增照片資料失敗: {e}") finally: db.close()
def get_all_photos(): db: Session = SessionLocal() try: photos = db.query(Photo).all() return photos except Exception as e: db.rollback() print(f"獲取照片資料失敗: {e}") finally: db.close()
def delete_photo_by_id(photo_id): db: Session = SessionLocal() try: db.query(Photo).filter(Photo.id == photo_id).delete() db.commit() print(f"刪除照片資料成功,ID: {photo_id}") except Exception as e: db.rollback() print(f"刪除照片資料失敗: {e}") finally: db.close()
add_photo( user_id="user_123", group_id="group_456", photo_name="example_photo.jpg", photo_url="http://example.com/photo.jpg", photo_thumbnail_url="http://example.com/photo_thumb.jpg", place_tmp="Taipei", photo_properties={"description": "A beautiful sunset."}, photo_image_hash="hash_value_example" )
photos = get_all_photos() print(photos)
delete_photo_by_id(1)
session = SessionLocal()
try: session.execute(text("DROP TABLE IF EXISTS photos_backup")) session.commit() except Exception as e: print(f"出錯了: {e}") session.rollback() finally: session.close()
|