Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_readonly_true(self, session_readonly):
# control exception when adding object to readonly gnucash db
v = Version(table_name="sample", table_version="other sample")
sa_session_readonly = session_readonly.session
sa_session_readonly.add(v)
with pytest.raises(GnucashException):
sa_session_readonly.commit()
# control exception when deleting object to readonly gnucash db
sa_session_readonly.delete(session_readonly.query(Account).first())
with pytest.raises(GnucashException):
sa_session_readonly.commit()
# control exception when modifying object to readonly gnucash db
sa_session_readonly.query(Account).first().name = "foo"
with pytest.raises(GnucashException):
sa_session_readonly.commit()
- CS = Income/Cap Gain (Short)/stock.mnemonic
- I = Income/Interest Income/stock.mnemonic
Args:
broker_account (:class:`piecash.core.account.Account`): the broker account where the account holding
the stock is to be created
income_account (:class:`piecash.core.account.Account`): the income account where the accounts holding
the income related to the stock are to be created
income_account_types (str): "/" separated codes to drive the creation of income accounts
Returns:
:class:`piecash.core.account.Account`: a tuple with the account under the broker_account where the stock is held
and the list of income accounts.
"""
if cdty.namespace == "CURRENCY":
raise GnucashException("{} is a currency ! You can't create stock_accounts for currencies".format(cdty))
from .account import Account
symbol = cdty.mnemonic
try:
acc = broker_account.children(name=symbol)
except KeyError:
acc = Account(symbol, "STOCK", cdty, broker_account)
inc_accounts = []
if income_account:
cur = cdty.base_currency
for inc_acc in income_account_types.split("/"):
sub_account_name = {
"D": "Dividend Income",
def readonly_commit(*args, **kwargs):
# session.rollback()
raise GnucashException("You cannot change the DB, it was opened as readonly!")
def prices_df(self):
"""
Return a pandas DataFrame with all prices (:class:`piecash.core.commodity.Price`) from the book
:return: :class:`pandas.DataFrame`
"""
try:
import pandas
except ImportError:
raise GnucashException("pandas is required to output dataframes")
# preload list of commodities
commodities = self.session.query(Commodity).all()
# load all prices
Currency = aliased(Commodity)
prices = self.session.query(Price) \
.join(Commodity, Price.commodity) \
.join(Currency, Price.currency) \
.order_by(Commodity.mnemonic, Price.date, Currency.mnemonic).all()
fields = ["date", "type", "value",
"commodity.guid", "commodity.mnemonic",
"currency.guid", "currency.mnemonic", ]
fields_getter = [attrgetter(fld) for fld in fields]
df_prices = pandas.DataFrame([[fg(pr) for fg in fields_getter]
)
if uri_conn == "sqlite:///:memory:":
raise ValueError("An in memory sqlite gnucash databook cannot be opened, it should be created")
# create database (if not sqlite in memory)
if not database_exists(uri_conn):
raise GnucashException("Database '{}' does not exist (please use create_book to create " \
"GnuCash books from scratch)".format(uri_conn))
engine = create_piecash_engine(uri_conn, **kwargs)
# backup database if readonly=False and do_backup=True
if not readonly and do_backup:
if engine.name != "sqlite":
raise GnucashException(
"Cannot do a backup for engine '{}'. Do yourself a backup and then specify do_backup=False".format(
engine.name))
url = uri_conn[len("sqlite:///"):].replace("?check_same_thread=False", "")
url_backup = url + ".{:%Y%m%d%H%M%S}.gnucash".format(datetime.datetime.now())
shutil.copyfile(url, url_backup)
locks = list(engine.execute(gnclock.select()))
# ensure the file is not locked by GnuCash itself
if locks and not open_if_lock:
raise GnucashException("Lock on the file")
s = Session(bind=engine)
def base_currency(self):
b = self.book
if b is None:
raise GnucashException("The commodity should be linked to a session to have a 'base_currency'")
if self.namespace == "CURRENCY":
# get the base currency as first commodity in DB
return b.default_currency
else:
# retrieve currency from quoted_currency kvp
# TODO: recover from the web (as fallback)
mnemonic = self.get("quoted_currency", None)
if mnemonic:
return b.currencies(mnemonic=mnemonic)
else:
raise GnucashException("The commodity '{}' has no information about its base currency. "
"Add a kvp item named 'quoted_currency' with the mnemonic of the "
"currency to have proper behavior".format(self.mnemonic))
import datetime
from decimal import Decimal
from sqlalchemy import Column, VARCHAR, INTEGER, ForeignKey, BIGINT, Index
from sqlalchemy.orm import relation
from sqlalchemy.orm.exc import MultipleResultsFound
from ._commodity_helper import quandl_fx
from .._common import CallableList, GncConversionError
from .._common import GnucashException, hybrid_property_gncnumeric
from .._declbase import DeclarativeBaseGuid
from ..sa_extra import _DateAsDateTime
from ..yahoo_client import get_latest_quote, download_quote
class GncCommodityError(GnucashException):
pass
class GncPriceError(GnucashException):
pass
class Price(DeclarativeBaseGuid):
"""
A single Price for a commodity.
Attributes:
commodity (:class:`Commodity`): commodity to which the Price relates
currency (:class:`Commodity`): currency in which the Price is expressed
date (:class:`datetime.date`): date object representing the day at which the price is relevant
source (str): source of the price
def base_currency(self):
b = self.book
if b is None:
raise GnucashException("The commodity should be linked to a session to have a 'base_currency'")
if self.namespace == "CURRENCY":
# get the base currency as first commodity in DB
return b.default_currency
else:
# retrieve currency from quoted_currency kvp
# TODO: recover from the web (as fallback)
mnemonic = self.get("quoted_currency", None)
if mnemonic:
return b.currencies(mnemonic=mnemonic)
else:
raise GnucashException("The commodity '{}' has no information about its base currency. "
"Add a kvp item named 'quoted_currency' with the mnemonic of the "
"currency to have proper behavior".format(self.mnemonic))
import locale
from decimal import Decimal
from sqlalchemy import Column, VARCHAR, INTEGER, cast, Float
from sqlalchemy.ext.hybrid import hybrid_property
from .sa_extra import DeclarativeBase, _Date
class GnucashException(Exception):
pass
class GncNoActiveSession(GnucashException):
pass
class GncValidationError(GnucashException):
pass
class GncImbalanceError(GncValidationError):
pass
class GncConversionError(GnucashException):
pass
class Recurrence(DeclarativeBase):
pass
class GncNoActiveSession(GnucashException):
pass
class GncValidationError(GnucashException):
pass
class GncImbalanceError(GncValidationError):
pass
class GncConversionError(GnucashException):
pass
class Recurrence(DeclarativeBase):
"""
Recurrence information for scheduled transactions
Attributes:
obj_guid (str): link to the parent ScheduledTransaction record.
recurrence_mult (int): Multiplier for the period type. Describes how many times
the period repeats for the next occurrence.
recurrence_period_type (str): type or recurrence (monthly, daily).
recurrence_period_start (date): the date the recurrence starts.
recurrence_weekend_adjust (str): adjustment to be made if the next occurrence
falls on weekend / non-working day.
"""