Black

piket_client/gui.py

         if dbus:
             try:
                 session_bus = dbus.SessionBus()
-                self.osk = session_bus.get_object('org.onboard.Onboard',
-                        '/org/onboard/Onboard/Keyboard')
+                self.osk = session_bus.get_object(
+                    "org.onboard.Onboard", "/org/onboard/Onboard/Keyboard"
+                )
             except dbus.exceptions.DBusException:
                 # Onboard not present or dbus broken
                 self.osk = None
         ag.setExclusive(True)
 
         for ct in ConsumptionType.get_all():
-            action = QAction(
-                self.load_icon(ct.icon or 'beer_bottle.svg'),
-                ct.name,
-                ag
-            )
+            action = QAction(self.load_icon(ct.icon or "beer_bottle.svg"), ct.name, ag)
             action.setCheckable(True)
             action.setData(str(ct.consumption_type_id))
 
         self.consumption_type_changed.connect(self.main_widget.consumption_type_changed)
         self.setCentralWidget(self.main_widget)
 
-
     @Slot(QAction)
     def consumption_type_change(self, action: QAction):
         self.consumption_type_changed.emit(action.data())
 
     def show_keyboard(self) -> None:
-        ''' Show the virtual keyboard, if possible. '''
+        """ Show the virtual keyboard, if possible. """
         if self.osk:
             self.osk.Show()
 
     def hide_keyboard(self) -> None:
-        ''' Hide the virtual keyboard, if possible. '''
+        """ Hide the virtual keyboard, if possible. """
         if self.osk:
             self.osk.Hide()
 

piket_client/logger.py

 from logging import getLogger
 
 logging.basicConfig(
-        format='%(asctime)s - %(name)s - %(levelname)s: %(message)s',
-        level=logging.INFO
+    format="%(asctime)s - %(name)s - %(levelname)s: %(message)s", level=logging.INFO
 )

piket_client/model.py

-'''
+"""
 Provides access to the models stored in the database, via the server.
-'''
+"""
 from typing import NamedTuple
 from urllib.parse import urljoin
 
 
 from . import logger
 
-LOG = logger.getLogger('model')
+LOG = logger.getLogger("model")
 
 SERVER_URL = "http://127.0.0.1:5000"
 
+
 class Person(NamedTuple):
-    ''' Represents a Person, as retrieved from the database. '''
+    """ Represents a Person, as retrieved from the database. """
+
     name: str
     person_id: int = None
     consumptions: dict = {}
 
     def add_consumption(self, type_id: str) -> bool:
-        ''' Register a consumption for this Person. '''
+        """ Register a consumption for this Person. """
         req = requests.post(
-            urljoin(SERVER_URL,
-                f'people/{self.person_id}/add_consumption/{type_id}')
+            urljoin(SERVER_URL, f"people/{self.person_id}/add_consumption/{type_id}")
         )
         try:
             data = req.json()
 
-            if 'error' in data:
+            if "error" in data:
                 LOG.error(
-                    'Could not add consumption for %s (%s): %s',
-                    self.person_id, req.status_code, data
+                    "Could not add consumption for %s (%s): %s",
+                    self.person_id,
+                    req.status_code,
+                    data,
                 )
                 return False
 
-            self.consumptions.update(data['person']['consumptions'])
+            self.consumptions.update(data["person"]["consumptions"])
 
             return True
         except ValueError:
             LOG.error(
-                'Did not get JSON on adding Consumption (%s): %s',
-                req.status_code, req.content
+                "Did not get JSON on adding Consumption (%s): %s",
+                req.status_code,
+                req.content,
             )
             return False
 
-    def create(self) -> 'Person':
-        ''' Create a new Person from the current attributes. As tuples are
-        immutable, a new Person with the correct id is returned. '''
+    def create(self) -> "Person":
+        """ Create a new Person from the current attributes. As tuples are
+        immutable, a new Person with the correct id is returned. """
         req = requests.post(
             urljoin(SERVER_URL, "people"), json={"person": {"name": self.name}}
         )
             data = req.json()
         except ValueError:
             LOG.error(
-                'Did not get JSON on adding Person (%s): %s',
-                req.status_code, req.content
+                "Did not get JSON on adding Person (%s): %s",
+                req.status_code,
+                req.content,
             )
             return None
 
-        if 'error' in data or req.status_code != 201:
-            LOG.error(
-                'Could not create Person (%s): %s',
-                req.status_code, data
-            )
+        if "error" in data or req.status_code != 201:
+            LOG.error("Could not create Person (%s): %s", req.status_code, data)
             return None
 
-        return Person.from_dict(data['person'])
+        return Person.from_dict(data["person"])
 
     @classmethod
-    def get(cls, person_id: int) -> 'Person':
-        ''' Retrieve a Person by id. '''
-        req = requests.get(urljoin(SERVER_URL, f'/people/{person_id}'))
+    def get(cls, person_id: int) -> "Person":
+        """ Retrieve a Person by id. """
+        req = requests.get(urljoin(SERVER_URL, f"/people/{person_id}"))
 
         try:
             data = req.json()
 
-            if 'error' in data:
+            if "error" in data:
                 LOG.warning(
-                    'Could not get person %s (%s): %s',
-                    person_id, req.status_code, data
+                    "Could not get person %s (%s): %s", person_id, req.status_code, data
                 )
                 return None
 
-            return Person.from_dict(data['person'])
+            return Person.from_dict(data["person"])
 
         except ValueError:
             LOG.error(
-                'Did not get JSON from server on getting Person (%s): %s',
-                req.status_code, req.content
+                "Did not get JSON from server on getting Person (%s): %s",
+                req.status_code,
+                req.content,
             )
             return None
 
     @classmethod
-    def get_all(cls) -> ['Person']:
-        ''' Get all active People. '''
-        req = requests.get(urljoin(SERVER_URL, '/people'))
+    def get_all(cls) -> ["Person"]:
+        """ Get all active People. """
+        req = requests.get(urljoin(SERVER_URL, "/people"))
 
         try:
             data = req.json()
 
-            if 'error' in data:
-                LOG.warning(
-                    'Could not get people (%s): %s',
-                    req.status_code, data
-                )
+            if "error" in data:
+                LOG.warning("Could not get people (%s): %s", req.status_code, data)
 
-            return [Person.from_dict(item) for item in data['people']]
+            return [Person.from_dict(item) for item in data["people"]]
 
         except ValueError:
             LOG.error(
-                'Did not get JSON from server on getting People (%s): %s',
-                req.status_code, req.content
+                "Did not get JSON from server on getting People (%s): %s",
+                req.status_code,
+                req.content,
             )
             return None
 
     @classmethod
-    def from_dict(cls, data: dict) -> 'Person':
-        ''' Reconstruct a Person object from a dict. '''
+    def from_dict(cls, data: dict) -> "Person":
+        """ Reconstruct a Person object from a dict. """
         return Person(
-            name = data['name'],
-            person_id = data['person_id'],
-            consumptions = data['consumptions']
+            name=data["name"],
+            person_id=data["person_id"],
+            consumptions=data["consumptions"],
         )
 
+
 class ConsumptionType(NamedTuple):
-    ''' Represents a stored ConsumptionType. '''
+    """ Represents a stored ConsumptionType. """
 
     name: str
     consumption_type_id: int = None
     icon: str = None
 
-    def create(self) -> 'Person':
-        ''' Create a new ConsumptionType from the current attributes. As tuples
+    def create(self) -> "Person":
+        """ Create a new ConsumptionType from the current attributes. As tuples
         are immutable, a new ConsumptionType with the correct id is returned.
-        '''
+        """
         req = requests.post(
-            urljoin(SERVER_URL, "consumption_types"), json={"consumption_type":
-                {"name": self.name, "icon": self.icon}}
+            urljoin(SERVER_URL, "consumption_types"),
+            json={"consumption_type": {"name": self.name, "icon": self.icon}},
         )
 
         try:
             data = req.json()
         except ValueError:
             LOG.error(
-                'Did not get JSON on adding ConsumptionType (%s): %s',
-                req.status_code, req.content
+                "Did not get JSON on adding ConsumptionType (%s): %s",
+                req.status_code,
+                req.content,
             )
             return None
 
-        if 'error' in data or req.status_code != 201:
+        if "error" in data or req.status_code != 201:
             LOG.error(
-                'Could not create ConsumptionType (%s): %s',
-                req.status_code, data
+                "Could not create ConsumptionType (%s): %s", req.status_code, data
             )
             return None
 
-        return Person.from_dict(data['consumption_type'])
+        return Person.from_dict(data["consumption_type"])
 
     @classmethod
-    def get(cls, consumption_type_id: int) -> 'ConsumptionType':
-        ''' Retrieve a ConsumptionType by id. '''
-        req = requests.get(urljoin(SERVER_URL,
-            f'/consumption_types/{consumption_type_id}'))
+    def get(cls, consumption_type_id: int) -> "ConsumptionType":
+        """ Retrieve a ConsumptionType by id. """
+        req = requests.get(
+            urljoin(SERVER_URL, f"/consumption_types/{consumption_type_id}")
+        )
 
         try:
             data = req.json()
 
-            if 'error' in data:
+            if "error" in data:
                 LOG.warning(
-                    'Could not get consumption type %s (%s): %s',
-                    consumption_type_id, req.status_code, data
+                    "Could not get consumption type %s (%s): %s",
+                    consumption_type_id,
+                    req.status_code,
+                    data,
                 )
                 return None
 
-            return cls.from_dict(data['consumption_type'])
+            return cls.from_dict(data["consumption_type"])
 
         except ValueError:
             LOG.error(
-                'Did not get JSON from server on getting consumption type (%s): %s',
-                req.status_code, req.content
+                "Did not get JSON from server on getting consumption type (%s): %s",
+                req.status_code,
+                req.content,
             )
             return None
 
     @classmethod
-    def get_all(cls) -> ['ConsumptionType']:
-        ''' Get all active ConsumptionTypes. '''
-        req = requests.get(urljoin(SERVER_URL, '/consumption_types'))
+    def get_all(cls) -> ["ConsumptionType"]:
+        """ Get all active ConsumptionTypes. """
+        req = requests.get(urljoin(SERVER_URL, "/consumption_types"))
 
         try:
             data = req.json()
 
-            if 'error' in data:
+            if "error" in data:
                 LOG.warning(
-                    'Could not get consumption types (%s): %s',
-                    req.status_code, data
+                    "Could not get consumption types (%s): %s", req.status_code, data
                 )
 
-            return [cls.from_dict(item) for item in data['consumption_types']]
+            return [cls.from_dict(item) for item in data["consumption_types"]]
 
         except ValueError:
             LOG.error(
-                'Did not get JSON from server on getting ConsumptionTypes (%s): %s',
-                req.status_code, req.content
+                "Did not get JSON from server on getting ConsumptionTypes (%s): %s",
+                req.status_code,
+                req.content,
             )
             return None
 
     @classmethod
-    def from_dict(cls, data: dict) -> 'ConsumptionType':
-        ''' Reconstruct a ConsumptionType from a dict. '''
+    def from_dict(cls, data: dict) -> "ConsumptionType":
+        """ Reconstruct a ConsumptionType from a dict. """
         return ConsumptionType(
-            name = data['name'],
-            consumption_type_id = data['consumption_type_id'],
-            icon = data.get('icon')
+            name=data["name"],
+            consumption_type_id=data["consumption_type_id"],
+            icon=data.get("icon"),
         )

piket_server/__init__.py

 
     return jsonify(person=person.as_dict, consumption=consumption.as_dict), 201
 
+
 @app.route("/people/<int:person_id>/add_consumption/<int:ct_id>", methods=["POST"])
 def add_consumption2(person_id: int, ct_id: int):
     person = Person.query.get_or_404(person_id)
 # ConsumptionType
 @app.route("/consumption_types", methods=["GET"])
 def get_consumption_types():
-    ''' Return a list of currently active consumption types. '''
+    """ Return a list of currently active consumption types. """
     ctypes = ConsumptionType.query.all()
     result = [ct.as_dict for ct in ctypes]
     return jsonify(consumption_types=result)
 
+
 @app.route("/consumption_types/<int:consumption_type_id>", methods=["GET"])
 def get_consumption_type(consumption_type_id: int):
     ct = ConsumptionType.query.get_or_404(consumption_type_id)
 
     return jsonify(consumption_type=ct.as_dict)
 
+
 @app.route("/consumption_types", methods=["POST"])
 def add_consumption_type():
     """ Add a new ConsumptionType.  """
     if not json:
         return jsonify({"error": "Could not parse JSON."}), 400
 
-    data = json.get('consumption_type') or {}
-    ct = ConsumptionType(
-        name=data.get('name'),
-        icon=data.get('icon')
-    )
+    data = json.get("consumption_type") or {}
+    ct = ConsumptionType(name=data.get("name"), icon=data.get("icon"))
 
     try:
         db.session.add(ct)
         db.session.commit()
     except SQLAlchemyError:
-        return jsonify({'error': 'Invalid arguments for ConsumptionType.'}), 400
+        return jsonify({"error": "Invalid arguments for ConsumptionType."}), 400
 
     return jsonify(consumption_type=ct.as_dict), 201

piket_server/alembic/env.py

 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
 import piket_server
+
 target_metadata = piket_server.db.Model.metadata
 
 # other values from the config, defined by the needs of env.py,
 # my_important_option = config.get_main_option("my_important_option")
 # ... etc.
 from piket_server import CONFIG_DIR, DB_URL
+
 os.makedirs(os.path.expanduser(CONFIG_DIR), mode=0o744, exist_ok=True)
 
-config.file_config['alembic']['sqlalchemy.url'] = DB_URL
+config.file_config["alembic"]["sqlalchemy.url"] = DB_URL
 
 
 def run_migrations_offline():
 
     """
     url = config.get_main_option("sqlalchemy.url")
-    context.configure(
-        url=url, target_metadata=target_metadata, literal_binds=True)
+    context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
 
     with context.begin_transaction():
         context.run_migrations()
     """
     connectable = engine_from_config(
         config.get_section(config.config_ini_section),
-        prefix='sqlalchemy.',
-        poolclass=pool.NullPool)
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
 
     with connectable.connect() as connection:
-        context.configure(
-            connection=connection,
-            target_metadata=target_metadata
-        )
+        context.configure(connection=connection, target_metadata=target_metadata)
 
         with context.begin_transaction():
             context.run_migrations()
 
+
 if context.is_offline_mode():
     run_migrations_offline()
 else:

piket_server/alembic/versions/491bb980d1d7_add_settlement_allow_null_person_on_.py

 
 
 # revision identifiers, used by Alembic.
-revision = '491bb980d1d7'
-down_revision = 'de101d627237'
+revision = "491bb980d1d7"
+down_revision = "de101d627237"
 branch_labels = None
 depends_on = None
 
 
 def upgrade():
     op.create_table(
-        'settlements',
-        sa.Column('settlement_id', sa.Integer, primary_key=True),
-        sa.Column('name', sa.String, nullable=False)
+        "settlements",
+        sa.Column("settlement_id", sa.Integer, primary_key=True),
+        sa.Column("name", sa.String, nullable=False),
     )
 
     op.add_column(
-        'consumptions',
+        "consumptions",
         sa.Column(
-            'settlement_id', sa.Integer,
-            sa.ForeignKey('settlements.settlement_id'),
-            nullable=True
-        )
+            "settlement_id",
+            sa.Integer,
+            sa.ForeignKey("settlements.settlement_id"),
+            nullable=True,
+        ),
     )
 
 
 def downgrade():
-    op.drop_column('consumptions', 'settlement_id')
+    op.drop_column("consumptions", "settlement_id")
 
-    op.drop_table('settlements')
+    op.drop_table("settlements")

piket_server/alembic/versions/a09086bfe84c_create_persons_table.py

 
 
 # revision identifiers, used by Alembic.
-revision = 'a09086bfe84c'
+revision = "a09086bfe84c"
 down_revision = None
 branch_labels = None
 depends_on = None
 
 def upgrade():
     op.create_table(
-        'people',
-        sa.Column('person_id', sa.Integer, primary_key=True),
-        sa.Column('name', sa.String, nullable=False)
+        "people",
+        sa.Column("person_id", sa.Integer, primary_key=True),
+        sa.Column("name", sa.String, nullable=False),
     )
 
 
 def downgrade():
-    op.drop_table('people')
+    op.drop_table("people")

piket_server/alembic/versions/de101d627237_create_consumptions_consumption_types.py

 
 
 # revision identifiers, used by Alembic.
-revision = 'de101d627237'
-down_revision = 'a09086bfe84c'
+revision = "de101d627237"
+down_revision = "a09086bfe84c"
 branch_labels = None
 depends_on = None
 
 
 def upgrade():
     op.create_table(
-        'consumption_types',
-        sa.Column('consumption_type_id', sa.Integer, nullable=False,
-            primary_key=True),
-        sa.Column('name', sa.String, nullable=False),
-        sa.Column('icon', sa.String, nullable=True),
+        "consumption_types",
+        sa.Column("consumption_type_id", sa.Integer, nullable=False, primary_key=True),
+        sa.Column("name", sa.String, nullable=False),
+        sa.Column("icon", sa.String, nullable=True),
     )
 
     op.create_table(
-        'consumptions',
-        sa.Column('consumption_id', sa.Integer, primary_key=True),
-        sa.Column('person_id', sa.Integer, nullable=True),
-        sa.Column('consumption_type_id', sa.Integer, nullable=False),
-        sa.Column('created_at', sa.DateTime, nullable=False),
-        sa.ForeignKeyConstraint(['consumption_type_id'], ['consumption_types.consumption_type_id'], ),
-        sa.ForeignKeyConstraint(['person_id'], ['people.person_id'], ),
-        sa.PrimaryKeyConstraint('consumption_id')
+        "consumptions",
+        sa.Column("consumption_id", sa.Integer, primary_key=True),
+        sa.Column("person_id", sa.Integer, nullable=True),
+        sa.Column("consumption_type_id", sa.Integer, nullable=False),
+        sa.Column("created_at", sa.DateTime, nullable=False),
+        sa.ForeignKeyConstraint(
+            ["consumption_type_id"], ["consumption_types.consumption_type_id"]
+        ),
+        sa.ForeignKeyConstraint(["person_id"], ["people.person_id"]),
+        sa.PrimaryKeyConstraint("consumption_id"),
     )
 
 
 def downgrade():
-    op.drop_table('consumptions')
-    op.drop_table('consumption_types')
+    op.drop_table("consumptions")
+    op.drop_table("consumption_types")