feat: faker data: query

app.py

@@ -3,8 +3,7 @@ import config
 import utils.response as response
 import service.faker_data as faker
 from base import db
-from model import Person, SexEnum, City, Hotel
-from sqlalchemy import select, join
+from model import SexEnum
 
 app = Flask(__name__)
 config.init_config(app)
@@ -12,9 +11,8 @@ db.init_app(app)
 
 
 @app.route('/')
-def hello_world():  # put application's code here
-    print(SexEnum(1).sex)
-    return 'Hello World!'
+def hello_world():
+    return f'<h1>nConTrack_Faker<h1>'
 
 
 @app.route("/faker/railway/<number>")
@@ -32,12 +30,30 @@ def faker_contacts(num_hotel, num_railway):
     return response.succ(faker.faker_contacts(int(num_hotel), int(num_railway)))
 
 
-@app.route("/test")
-def test():
-    print(
-        select(Hotel.identification).
-        join(Hotel, )
-    )
+@app.route("/query/patient")
+def query_patient():
+    return response.succ(faker.query_patients())
+
+
+@app.route("/query/contacts/hotel/<identification>")
+def query_contacts_hotel(identification):
+    return response.succ(faker.query_contacts_hotel(identification))
+
+
+@app.route("/query/contacts/railway/<identification>")
+def query_contacts_railway(identification):
+    return response.succ(faker.query_contacts_railway(identification))
+
+
+@app.route("/query/contacts/classify/<identification>")
+def query_contacts_classify(identification):
+    return response.succ(faker.query_contacts_classify(identification))
+
+
+@app.route("/query/contacts/<identification>")
+def query_contacts(identification):
+    return response.succ(faker.query_contacts(identification))
+
 
 if __name__ == '__main__':
     app.run()

config.py

@@ -1,5 +1,3 @@
-from urllib.parse import quote_plus as urlquote
-
 DIALECT = 'mysql'
 DRIVER = 'pymysql'
 USERNAME = 'root'

model/city.py

@@ -18,3 +18,6 @@ class City(db.Model):
     latitude = Column(String(255))
     longitude = Column(String(255))
     ad_code = Column(String(255))
+
+    def __to_dict__(self) -> dict:
+        return {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}

model/hotel.py

@@ -12,3 +12,6 @@ class Hotel(db.Model):
     identification = Column(String(18), server_default=text("''"), comment='入住者身份证号')
     in_data = Column(Date, nullable=False, comment='入住日期')
     out_data = Column(Date, nullable=False, comment='离开日期')
+
+    def __to_dict__(self) -> dict:
+        return {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}

model/railway.py

@@ -11,3 +11,6 @@ class Railway(db.Model):
     phone = Column(String(20), nullable=False, server_default=text("''"), comment='手机号码')
     train = Column(String(10), nullable=False, server_default=text("''"), comment='车次')
     launch = Column(Date, nullable=False, comment='发车时间')
+
+    def __to_dict__(self) -> dict:
+        return {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}

service/faker_data.py

@@ -134,21 +134,17 @@ def faker_contacts(num_hotel: int, num_railway: int):
     persons = faker_identifications(num_railway + num_hotel + 1)
     patient = persons[0]
     contacts_hotel = persons[1:1 + num_hotel]
-    contacts_railway = persons[-1-num_railway: -1]
-    print(persons)
-    print(patient)
-    print(contacts_railway)
-    print(contacts_hotel)
+    contacts_railway = persons[0 - num_railway:]
     hotel_patient = faker_hotel(patient)
     railway_patient = faker_railway(patient)
 
     def assemble_hotel(contact: Person) -> Hotel:
-        hotel_contact = copy.copy(hotel_patient)
+        hotel_contact = copy.deepcopy(hotel_patient)
         hotel_contact.identification = contact.identification
         return hotel_contact
 
     def assemble_railway(contact: Person) -> Railway:
-        railway_contact = copy.copy(railway_patient)
+        railway_contact = copy.deepcopy(railway_patient)
         railway_contact.identification = contact.identification
         railway_contact.name = contact.name
         return railway_contact
@@ -171,3 +167,45 @@ def faker_contacts(num_hotel: int, num_railway: int):
         }
     }
 
+
+def query_patients() -> list[dict]:
+    patients = db.session.query(Person) \
+        .join(Hotel, Person.identification == Hotel.identification) \
+        .join(Railway, Person.identification == Railway.identification) \
+        .all()
+    return [patient.__to_dict__() for patient in patients]
+
+
+def query_contacts_hotel(identification: str) -> list[dict]:
+    patient_subquery = Hotel.query.filter(Hotel.identification == identification).subquery()
+    contacts = db.session.query(Hotel, Person) \
+        .join(Hotel, Hotel.identification == Person.identification) \
+        .filter(Person.identification != identification) \
+        .filter(Hotel.out_data > patient_subquery.c.in_data)
+    return [merge_object2dict(hotel, person) for hotel, person in contacts]
+
+
+def query_contacts_railway(identification: str) -> list[dict]:
+    patient_subquery = Railway.query.filter(Railway.identification == identification).subquery()
+    contacts = db.session.query(Railway, Person) \
+        .join(Railway, Railway.identification == Person.identification) \
+        .filter(Person.identification != identification) \
+        .filter(Railway.launch == patient_subquery.c.launch)
+    return [merge_object2dict(railway, person) for railway, person in contacts]
+
+
+def query_contacts_classify(identification: str) -> dict[str, list[dict]]:
+    contacts = {'hotel': query_contacts_hotel(identification), 'railway': query_contacts_railway(identification)}
+    return contacts
+
+
+def query_contacts(identification: str) -> list[dict]:
+    contacts = [*query_contacts_hotel(identification), *query_contacts_railway(identification)]
+    return contacts
+
+
+def merge_object2dict(*objs: db.Model) -> dict:
+    res = {}
+    for obj in objs:
+        res.update(obj.__to_dict__())
+    return res