diff --git a/README.md b/README.md index f47eaa3..73bab63 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,7 @@ from actual.queries import get_accounts, create_transaction_from_ids with Actual(base_url="http://localhost:5006", password="mypass", file="My budget") as actual: act = get_accounts(actual.session)[0] # get first account t = create_transaction_from_ids( - actual.session, act.id, datetime.date.today(), None, notes="My first transaction", amount=decimal.Decimal(10.5) + actual.session, datetime.date.today(), act.id, None, notes="My first transaction", amount=decimal.Decimal(10.5) ) actual.session.add(t) actual.commit() # use the actual.commit() instead of session.commit()! diff --git a/actual/database.py b/actual/database.py index 245ee3f..358164b 100644 --- a/actual/database.py +++ b/actual/database.py @@ -94,6 +94,11 @@ def convert(self, is_new: bool = True) -> List[Message]: changes.append(m) return changes + def delete(self): + if not getattr(self, "tombstone", None): + raise AttributeError(f"Model {self.__name__} has no tombstone field and cannot be deleted.") + setattr(self, "tombstone", 1) + class Meta(SQLModel, table=True): __tablename__ = "__meta__" @@ -130,14 +135,14 @@ class Accounts(BaseModel, table=True): transactions: List["Transactions"] = Relationship(back_populates="account") -class Banks(SQLModel, table=True): +class Banks(BaseModel, table=True): id: Optional[str] = Field(default=None, sa_column=Column("id", Text, primary_key=True)) bank_id: Optional[str] = Field(default=None, sa_column=Column("bank_id", Text)) name: Optional[str] = Field(default=None, sa_column=Column("name", Text)) tombstone: Optional[int] = Field(default=None, sa_column=Column("tombstone", Integer, server_default=text("0"))) -class Categories(SQLModel, table=True): +class Categories(BaseModel, table=True): hidden: bool = Field(sa_column=Column("hidden", Boolean, nullable=False, server_default=text("0"))) id: Optional[str] = Field(default=None, sa_column=Column("id", Text, primary_key=True)) name: Optional[str] = Field(default=None, sa_column=Column("name", Text)) @@ -153,7 +158,7 @@ class Categories(SQLModel, table=True): group: "CategoryGroups" = Relationship(back_populates="categories", sa_relationship_kwargs={"uselist": False}) -class CategoryGroups(SQLModel, table=True): +class CategoryGroups(BaseModel, table=True): __tablename__ = "category_groups" hidden: bool = Field(sa_column=Column("hidden", Boolean, nullable=False, server_default=text("0"))) @@ -166,7 +171,7 @@ class CategoryGroups(SQLModel, table=True): categories: List["Categories"] = Relationship(back_populates="group") -class CategoryMapping(SQLModel, table=True): +class CategoryMapping(BaseModel, table=True): __tablename__ = "category_mapping" id: Optional[str] = Field(default=None, sa_column=Column("id", Text, primary_key=True)) @@ -215,14 +220,14 @@ class Notes(SQLModel, table=True): note: Optional[str] = Field(default=None, sa_column=Column("note", Text)) -class PayeeMapping(SQLModel, table=True): +class PayeeMapping(BaseModel, table=True): __tablename__ = "payee_mapping" id: Optional[str] = Field(default=None, sa_column=Column("id", Text, primary_key=True)) target_id: Optional[str] = Field(default=None, sa_column=Column("targetId", Text)) -class Payees(SQLModel, table=True): +class Payees(BaseModel, table=True): id: Optional[str] = Field(default=None, sa_column=Column("id", Text, primary_key=True)) name: Optional[str] = Field(default=None, sa_column=Column("name", Text)) category: Optional[str] = Field(default=None, sa_column=Column("category", Text)) @@ -246,7 +251,7 @@ class ReflectBudgets(SQLModel, table=True): goal: Optional[int] = Field(default=None, sa_column=Column("goal", Integer, server_default=text("null"))) -class Rules(SQLModel, table=True): +class Rules(BaseModel, table=True): id: Optional[str] = Field(default=None, sa_column=Column("id", Text, primary_key=True)) stage: Optional[str] = Field(default=None, sa_column=Column("stage", Text)) conditions: Optional[str] = Field(default=None, sa_column=Column("conditions", Text)) @@ -293,7 +298,7 @@ class SchedulesNextDate(SQLModel, table=True): tombstone: Optional[int] = Field(default=None, sa_column=Column("tombstone", Integer, server_default=text("0"))) -class TransactionFilters(SQLModel, table=True): +class TransactionFilters(BaseModel, table=True): __tablename__ = "transaction_filters" id: Optional[str] = Field(default=None, sa_column=Column("id", Text, primary_key=True)) diff --git a/actual/queries.py b/actual/queries.py index 0275adf..a5d2ce6 100644 --- a/actual/queries.py +++ b/actual/queries.py @@ -17,6 +17,7 @@ Payees, Transactions, ) +from actual.exceptions import ActualError T = typing.TypeVar("T") @@ -50,7 +51,23 @@ def is_uuid(text: str, version: int = 4): return False -def get_transactions(s: Session, notes: str = None, include_deleted: bool = False) -> typing.List[Transactions]: +def get_transactions( + s: Session, + start_date: datetime.date = None, + end_date: datetime.date = None, + notes: str = None, + include_deleted: bool = False, +) -> typing.List[Transactions]: + """ + Returns a list of all available transactions. + + :param s: session from Actual local database. + :param start_date: optional start date for the transaction period (inclusive) + :param end_date: optional end date for the transaction period (exclusive) + :param notes: optional notes filter for the transactions, case-insensitive. + :param include_deleted: includes deleted transactions from the search + :return: list of transactions with `account`, `category` and `payee` pre-loaded. + """ query = ( s.query(Transactions) .options( @@ -70,6 +87,10 @@ def get_transactions(s: Session, notes: str = None, include_deleted: bool = Fals Transactions.id, ) ) + if start_date: + query = query.filter(Transactions.date >= int(datetime.date.strftime(start_date, "%Y%m%d"))) + if end_date: + query = query.filter(Transactions.date < int(datetime.date.strftime(end_date, "%Y%m%d"))) if not include_deleted: query = query.filter(sqlalchemy.func.coalesce(Transactions.tombstone, 0) == 0) if notes: @@ -79,13 +100,14 @@ def get_transactions(s: Session, notes: str = None, include_deleted: bool = Fals def create_transaction_from_ids( s: Session, - account_id: str, date: datetime.date, + account_id: str, payee_id: typing.Optional[str], notes: str, category_id: str = None, amount: decimal.Decimal = 0, ) -> Transactions: + """Internal method to generate a transaction from ids instead of objects.""" date_int = int(datetime.date.strftime(date, "%Y%m%d")) t = Transactions( id=str(uuid.uuid4()), @@ -105,23 +127,38 @@ def create_transaction_from_ids( def create_transaction( s: Session, - account_name: str, date: datetime.date, - payee_name: str | Payees, - notes: str, - category_name: str = None, - amount: decimal.Decimal = 0, -): - acct = get_account(s, account_name) - payee = get_or_create_payee(s, payee_name) - if category_name: - category_id = get_or_create_category(s, category_name, "").id + account: str | Accounts, + payee: str | Payees, + notes: str = "", + category: str | Categories = None, + amount: decimal.Decimal | float | int = 0, +) -> Transactions: + """ + Creates a transaction from the provided input. + + :param s: session from Actual local database. + :param date: date of the transaction. + :param account: either account name or account object (via `get_account` or `get_accounts`). Will not be + auto-created if missing. + :param payee: name of the payee from the transaction. Will be created if missing. + :param notes: optional description for the transaction. + :param category: optional category for the transaction. Will be created if not existing. + :param amount: amount of the transaction. Positive indicates that the account balance will go up (deposit), and + negative that the account balance will go down (payment) + :return: the generated transaction object. + """ + acct = get_account(s, account) + payee = get_or_create_payee(s, payee) + if category: + category_id = get_or_create_category(s, category, "").id else: category_id = None - return create_transaction_from_ids(s, acct.id, date, payee.id, notes, category_id, amount) + return create_transaction_from_ids(s, date, acct.id, payee.id, notes, category_id, amount) def base_query(s: Session, instance: typing.Type[T], name: str, include_deleted: bool = False) -> typing.List[T]: + """Internal method to reduce querying complexity on sub-functions.""" query = s.query(instance) if not include_deleted: query = query.filter(sqlalchemy.func.coalesce(instance.tombstone, 0) == 0) @@ -131,19 +168,33 @@ def base_query(s: Session, instance: typing.Type[T], name: str, include_deleted: def create_category_group(s: Session, name: str) -> CategoryGroups: + """Creates a new category with the group name `name`. Make sure you avoid creating payees with duplicate names, as + it makes it difficult to find them without knowing the unique id beforehand.""" category_group = CategoryGroups(id=str(uuid.uuid4()), name=name, is_income=0, is_hidden=0, sort_order=0) s.add(category_group) return category_group def get_or_create_category_group(s: Session, name: str) -> CategoryGroups: - category_group = s.query(CategoryGroups).filter(CategoryGroups.name == name).one_or_none() + """Gets or create the category group, if not found with `name`. Deleted category groups are excluded from the + search.""" + category_group = ( + s.query(CategoryGroups).filter(CategoryGroups.name == name, CategoryGroups.tombstone == 0).one_or_none() + ) if not category_group: category_group = create_category_group(s, name) return category_group def get_categories(s: Session, name: str = None, include_deleted: bool = False) -> typing.List[Categories]: + """ + Returns a list of all available categories. + + :param s: session from Actual local database. + :param name: pattern name of the payee, case-insensitive. + :param include_deleted: includes all payees which were deleted via frontend. They would not show normally. + :return: list of categories with `transactions` already loaded. + """ query = base_query(s, Categories, name, include_deleted).options(joinedload(Payees.transactions)) return query.all() @@ -151,9 +202,16 @@ def get_categories(s: Session, name: str = None, include_deleted: bool = False) def create_category( s: Session, name: str, - group_name: str, + group_name: str = None, ) -> Categories: - category_group = get_or_create_category_group(s, group_name) + """Creates a new category with the `name` and `group_name`. If the group is not existing, it will also be created. + Make sure you avoid creating categories with duplicate names, as it makes it difficult to find them without knowing + the unique id beforehand. The exception is to have them in separate group names, but you then need to provide the + group name to the method also. + + If a group name is not provided, the default 'Usual Expenses' will be picked. + """ + category_group = get_or_create_category_group(s, group_name if group_name is not None else "Usual Expenses") category = Categories( id=str(uuid.uuid4()), name=name, hidden=0, is_income=0, sort_order=0, cat_group=category_group.id ) @@ -166,12 +224,13 @@ def create_category( def get_category( s: Session, name: str | Categories, group_name: str = None, strict_group: bool = False ) -> typing.Optional[Categories]: + """Gets an existing category by name, returns `None` if not found. Deleted payees are excluded from the search.""" if isinstance(name, Categories): return name category = ( s.query(Categories) .join(CategoryGroups) - .filter(Categories.name == name, CategoryGroups.name == group_name) + .filter(Categories.name == name, Categories.tombstone == 0, CategoryGroups.name == group_name) .one_or_none() ) if not category and not strict_group: @@ -181,8 +240,13 @@ def get_category( def get_or_create_category( - s: Session, name: str | Categories, group_name: str, strict_group: bool = False + s: Session, name: str | Categories, group_name: str = None, strict_group: bool = False ) -> Categories: + """Gets or create the category, if not found with `name`. If the category already exists, but in a different group, + but the category name is still unique, it will be returned, unless `strict_group` is set to `True`. + + If a group name is not provided, the default 'Usual Expenses' will be picked. + """ category = get_category(s, name, group_name, strict_group) if not category: category = create_category(s, name, group_name) @@ -190,22 +254,41 @@ def get_or_create_category( def get_accounts(s: Session, name: str = None, include_deleted: bool = False) -> typing.List[Accounts]: + """ + Returns a list of all available accounts. + + :param s: session from Actual local database. + :param name: pattern name of the payee, case-insensitive. + :param include_deleted: includes all payees which were deleted via frontend. They would not show normally. + :return: list of accounts with `transactions` already loaded. + """ query = base_query(s, Accounts, name, include_deleted).options(joinedload(Accounts.transactions)) return query.all() def get_payees(s: Session, name: str = None, include_deleted: bool = False) -> typing.List[Payees]: + """ + Returns a list of all available payees. + + :param s: session from Actual local database. + :param name: pattern name of the payee, case-insensitive. + :param include_deleted: includes all payees which were deleted via frontend. They would not show normally. + :return: list of payees with `transactions` already loaded. + """ query = base_query(s, Payees, name, include_deleted).options(joinedload(Payees.transactions)) return query.all() -def get_payee(s: Session, name: str | Payees) -> Payees: +def get_payee(s: Session, name: str | Payees) -> typing.Optional[Payees]: + """Gets an existing payee by name, returns `None` if not found. Deleted payees are excluded from the search.""" if isinstance(name, Payees): return name - return s.query(Payees).filter(Payees.name == name).one_or_none() + return s.query(Payees).filter(Payees.name == name, Payees.tombstone == 0).one_or_none() def create_payee(s: Session, name: str | None) -> Payees: + """Creates a new payee with the desired name. Make sure you avoid creating payees with duplicate names, as it makes + it difficult to find them without knowing the unique id beforehand.""" payee = Payees(id=str(uuid.uuid4()), name=name) s.add(payee) # add also the payee mapping @@ -214,6 +297,8 @@ def create_payee(s: Session, name: str | None) -> Payees: def get_or_create_payee(s: Session, name: str | Payees | None) -> Payees: + """Gets an existing payee by name, and if it does not exist, creates a new one. If the payee is created twice, + this method will fail with a database error.""" payee = get_payee(s, name) if not payee: payee = create_payee(s, name) @@ -223,6 +308,8 @@ def get_or_create_payee(s: Session, name: str | Payees | None) -> Payees: def create_account( s: Session, name: str, initial_balance: decimal.Decimal = decimal.Decimal(0), off_budget: bool = False ) -> Accounts: + """Creates a new account with the name and balance. Make sure you avoid creating accounts with duplicate names, as + it makes it difficult to find them without knowing the unique id beforehand.""" acct = Accounts(id=str(uuid.uuid4()), name=name, offbudget=int(off_budget), closed=0) s.add(acct) # add a blank payee @@ -234,22 +321,27 @@ def create_account( payee_starting = get_or_create_payee(s, "Starting Balance") category = get_or_create_category(s, "Starting Balances", "Income") create_transaction_from_ids( - s, acct.id, datetime.date.today(), payee_starting.id, "", category.id, initial_balance + s, datetime.date.today(), acct.id, payee_starting.id, "", category.id, initial_balance ) return acct def get_account(s: Session, name: str | Accounts) -> typing.Optional[Accounts]: + """ + Gets an account with the desired name, otherwise returns `None`. Deleted accounts are excluded from the search. + """ if isinstance(name, Accounts): return name if is_uuid(name): - account = s.query(Accounts).filter(Accounts.id == name).one_or_none() + account = s.query(Accounts).filter(Accounts.id == name, Accounts.tombstone == 0).one_or_none() else: - account = s.query(Accounts).filter(Accounts.name == name).one_or_none() + account = s.query(Accounts).filter(Accounts.name == name, Accounts.tombstone == 0).one_or_none() return account def get_or_create_account(s: Session, name: str | Accounts) -> Accounts: + """Gets or create the account, if not found with `name`. The initial balance will be set to 0 if an account is + created using this method.""" account = get_account(s, name) if not account: account = create_account(s, name) @@ -258,16 +350,31 @@ def get_or_create_account(s: Session, name: str | Accounts) -> Accounts: def create_transfer( s: Session, + date: datetime.date, source_account: str | Accounts, dest_account: str | Accounts, - amount: decimal.Decimal, - date: datetime.date, + amount: decimal.Decimal | int | float, notes: str = None, ) -> typing.Tuple[Transactions, Transactions]: + """ + Creates a transfer of money between two accounts, from `source_account` to `dest_account`. The amount is provided + as a positive value. + + :param s: session from Actual local database. + :param date: date of the transfer. + :param source_account: account that will transfer the money, and reduce its balance. + :param dest_account: account that will receive the money, and increase its balance. + :param amount: amount, as a positive decimal, to be transferred. + :param notes: additional description for the transfer. + :return: tuple containing both transactions, as one is created per account. The transactions would be + cross-referenced by their `transferred_id`. + """ + if amount <= 0: + raise ActualError("Amount must be a positive value.") source: Accounts = get_account(s, source_account) dest: Accounts = get_account(s, dest_account) - source_transaction = create_transaction_from_ids(s, source.id, date, dest.payee.id, notes, None, -amount) - dest_transaction = create_transaction_from_ids(s, dest.id, date, source.payee.id, notes, None, amount) + source_transaction = create_transaction_from_ids(s, date, source.id, dest.payee.id, notes, None, -amount) + dest_transaction = create_transaction_from_ids(s, date, dest.id, source.payee.id, notes, None, amount) # swap the transferred ids source_transaction.transferred_id = dest_transaction.id dest_transaction.transferred_id = source_transaction.id