Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- """Database module."""
- import json
- from datetime import datetime
- from sqlalchemy import Column, DateTime, Float, ForeignKey, Integer, String
- from sqlalchemy.orm import relationship
- from main import db
- class City(db.Model):
- """City table."""
- __tablename__ = "city"
- # __table_args__ = {'extend_existing': True}
- id = Column(Integer(), primary_key=True, autoincrement=True)
- name = Column(String(100), nullable=False)
- country = Column(String(2), nullable=False)
- class Service(db.Model):
- """Weather service table."""
- __tablename__ = "service"
- # __table_args__ = {'extend_existing': True}
- id = Column(Integer(), primary_key=True, autoincrement=True)
- name = Column(String(100), nullable=False)
- url = Column(String(250), nullable=False)
- class Weather(db.Model):
- """Weather table."""
- __tablename__ = "weather_city_service"
- # __table_args__ = {'extend_existing': True}
- # # __abstract__ = True
- city = Column(Integer, ForeignKey("city.id"), primary_key=True, nullable=False)
- service = Column(Integer, ForeignKey("service.id"), primary_key=True, nullable=False)
- temp = Column(Float(2), nullable=False)
- dt = Column(DateTime(), nullable=False)
- city_rel = relationship("City")
- service_rel = relationship("Service")
- def __init__(self, temp: float, city: int, service: int, dt: DateTime = None, id: int = None):
- self.id = id
- self.temp = temp
- self.dt = datetime.fromtimestamp(dt) if dt else datetime.utcnow()
- self.city = city
- self.service = service if isinstance(service, int) else self.__set_service(service)
- date_time = property()
- @date_time.setter
- def date_time(self, timetuple):
- if isinstance(timetuple, int):
- self.dt = datetime.fromtimestamp(timetuple)
- @staticmethod
- def __set_service(name):
- service = Service.query.filter_by(name=name).first()
- return service.id
- if __name__ == "__main__":
- db.drop_all()
- db.create_all()
- data_to_model = {"Service": Service, "City": City}
- with open("inits.json", "r", encoding="utf-8") as f:
- file = json.load(f)
- entities = [data_to_model[dic["model"]](**dic["values"]) for dic in file]
- db.session.bulk_save_objects(entities)
- db.session.commit()
- print("Initial inserts done")
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement