"""Database models.""" import os import typing import sqlalchemy import sqlalchemy.orm.decl_api from sqlalchemy import func from sqlalchemy.dialects import postgresql from sqlalchemy.ext.associationproxy import association_proxy from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.orderinglist import ordering_list from sqlalchemy.orm import relationship from sqlalchemy.schema import Column, ForeignKey from sqlalchemy.types import Boolean, Date, DateTime, Integer, String from .database import session Base: sqlalchemy.orm.decl_api.DeclarativeMeta = declarative_base() Base.query = session.query_property() class TimeStampedModel(Base): """Time stamped model.""" __abstract__ = True created = Column(DateTime, default=func.now()) modified = Column(DateTime, default=func.now(), onupdate=func.now()) class Series(TimeStampedModel): """Conference series.""" __tablename__ = "series" id = Column(Integer, primary_key=True) name = Column(String, nullable=False) slug = Column(String, unique=True) wikidata_qid = Column(String, unique=True) conferences = relationship("Conference", back_populates="series") class Conference(TimeStampedModel): """Conference.""" __tablename__ = "conference" id = Column(Integer, primary_key=True) title = Column(String, nullable=False) start = Column(Date) end = Column(Date) days = Column(Integer) timezone = Column(String) location = Column(String) country = Column(String) acronym = Column(String) url = Column(String) schedule_xml_url = Column(String) short_name = Column(String, unique=True) venue_id = Column(Integer, ForeignKey("venue.id")) online = Column(Boolean) wikidata_qid = Column(String, unique=True) series_id = Column(Integer, ForeignKey("series.id")) people_detail = relationship( "ConferencePerson", lazy="dynamic", back_populates="conference" ) people = association_proxy("people_detail", "person") events = relationship( "Event", order_by="Event.event_date", back_populates="conference", lazy="dynamic", ) venue = relationship("Venue", back_populates="conferences") series = relationship("Series", back_populates="conferences") class City(TimeStampedModel): """City.""" __tablename__ = "city" id = Column(Integer, primary_key=True) slug = Column(String, nullable=False, unique=True) name = Column(String, nullable=False) country_code = Column(String, ForeignKey("country.alpha2"), nullable=False) wikidata_qid = Column(String, nullable=False) venues = relationship("Venue", back_populates="city") country = relationship("Country", back_populates="cities") class Venue(TimeStampedModel): """Venue.""" __tablename__ = "venue" id = Column(Integer, primary_key=True) name = Column(String, nullable=False) city_id = Column(Integer, ForeignKey("city.id")) wikidata_qid = Column(String, nullable=False) conferences = relationship("Conference", back_populates="venue") city = relationship("City", back_populates="venues") class Country(TimeStampedModel): """Country.""" __tablename__ = "country" alpha2 = Column(String, primary_key=True) name = Column(String) wikidata_qid = Column(String, nullable=False) cities = relationship("City", back_populates="country") @property def flag(self) -> str: a = ord("A") flag_a = 0x1F1E6 char1, char2 = ( flag_a + ord(self.alpha2[0]) - a, flag_a + ord(self.alpha2[1]) - a, ) return chr(char1) + chr(char2) class ConferencePerson(Base): __tablename__ = "conference_person" conference_id = Column(Integer, ForeignKey("conference.id"), primary_key=True) person_id = Column(Integer, ForeignKey("person.id"), primary_key=True) named_as = Column(String) bio = Column(String) slug = Column(String) url = Column(String) affiliation = Column(String) photo_url = Column(String) person = relationship("Person", back_populates="conferences_association") conference = relationship("Conference", back_populates="people_detail") @property def events(self): return ( Event.query.join(EventPerson) .filter( Event.conference == self.conference, EventPerson.person == self.person ) .order_by(Event.event_date.desc()) ) class Event(TimeStampedModel): """Event.""" __tablename__ = "event" id = Column(Integer, primary_key=True) conference_id = Column(Integer, ForeignKey("conference.id"), nullable=False) event_date = Column(DateTime) # day = Column(Integer) guid = Column(String) start = Column(String) duration = Column(String) room = Column(String) track = Column(String) slug = Column(String) title = Column(String, nullable=False) abstract = Column(String) description = Column(String) event_type = Column(String) url = Column(String) cancelled = Column(Boolean) conference = relationship("Conference", back_populates="events") people_detail = relationship( "EventPerson", order_by="EventPerson.position", lazy="dynamic", back_populates="event", collection_class=ordering_list("position"), ) people = association_proxy( "people_detail", "person", creator=lambda i: EventPerson(person=i), ) class Person(TimeStampedModel): """Person.""" __tablename__ = "person" id = Column(Integer, primary_key=True) name = Column(String) wikidata_qid = Column(String) gender = Column(String) wikidata_photo = Column(postgresql.ARRAY(String)) events_association = relationship( "EventPerson", back_populates="person", lazy="dynamic", ) events = association_proxy("events_association", "event") conferences_association = relationship( "ConferencePerson", lazy="dynamic", back_populates="person" ) conferences = association_proxy("conferences_association", "conference") @property def conference_count(self): return ConferencePerson.query.filter_by(person_id=self.id).count() @property def event_count(self): return EventPerson.query.filter_by(person_id=self.id).count() def active_years(self): q = ( session.query(func.min(Event.event_date), func.max(Event.event_date)) .join(EventPerson) .filter_by(person_id=self.id) ) return q.one() # photos = relationship("PersonPhoto", back_populates="person") def events_by_time(self): q = ( session.query(Event) .join(EventPerson) .filter(EventPerson.person == self) .order_by(Event.event_date.desc()) ) return q def conference_by_time(self): q = ( session.query(ConferencePerson) .join(Conference) .filter(ConferencePerson.person == self) .order_by(Conference.start.desc()) ) return q def bio_source(self) -> ConferencePerson | None: bio_list = [cp for cp in self.conferences_association if cp.bio] if not bio_list: return None if len(bio_list) == 1: return typing.cast(ConferencePerson, bio_list[0]) recent = max(bio_list, key=lambda cp: cp.conference.start) len_recent_bio = len(recent.bio) longest = max(bio_list, key=lambda cp: len(cp.bio)) if recent == longest: return typing.cast(ConferencePerson, recent) best = longest if len(longest.bio) > len_recent_bio * 2 else recent return typing.cast(ConferencePerson, best) def photo_filename(self) -> str | None: if self.wikidata_photo: return os.path.join("wikidata_photo", "thumb", self.wikidata_photo[0]) q = self.conferences_association.filter(ConferencePerson.photo_url.isnot(None)) if q.count() == 0: return None best = max(q, key=lambda cp: cp.conference.start) ext = best.photo_url.rpartition(".")[-1] filename = f"{best.conference_id}_{self.id}.{ext}" return os.path.join("conference_photo", filename) # class PersonPhoto(TimeStampedModel): # """Person photo.""" # # __tablename__ = "person_photo" # person_id = Column(Integer, ForeignKey("person.id"), primary_key=True) # source_filename = Column(String) # # person = relationship("Person", back_populates="photos") class EventPerson(Base): """Event person.""" __tablename__ = "event_person" event_id = Column(Integer, ForeignKey("event.id"), primary_key=True) person_id = Column(Integer, ForeignKey("person.id"), primary_key=True) position = Column(Integer, nullable=False) person = relationship("Person", back_populates="events_association") event = relationship("Event", back_populates="people_detail")