gn2pg package

Submodules

gn2pg.api module

Provide python interface to GeoNature API.

Methods, see each class

Properties:

  • transfer_errors - Return number of HTTP errors

Exceptions:

  • APIException - General exception

  • HTTPError - HTTP protocol error

exception gn2pg.api.APIException

Bases: Exception

An exception occurred while handling your request.

class gn2pg.api.BaseAPI(config: Gn2PgSourceConf, controler: str)

Bases: object

Top class, not for direct use. Provides internal and template methods to use GeoNature API.

property controler: str | None

Return the controler name.

get_page(page_url: str) → dict | None

Get data from one API page

Parameters:

page_url (str) – page URL

Returns:

Datas as dict

Return type:

dict

property http_status: int

Return the latest HTTP status code.

page_list(params: dict, kind: str = 'data', pagination_param: str = 'offset') → tuple[List[str] | None, int]

List offset pages to download data, based on API “total_filtered” and “limit” values

Parameters:

  • params (dict) – Querystrings

  • kind (str, optional) – kind of data, defaults to “data”

  • pagination_param (str, optional) – Pagination parameter key

Returns:

url page list

Return type:

Optional[List[str]]

property transfer_errors: int

Return the number of HTTP errors during this session.

property version: str

Return version.

class gn2pg.api.DataAPI(config)

Bases: BaseAPI

Data API

exception gn2pg.api.ExportModuleNotFoundError

Bases: Exception

Custom exception raised when the EXPORTS module is not found.

gn2pg.check_conf module

TOML validation tools

class gn2pg.check_conf.Db(host: str, user: str, password: str, name: str, port: int = 5432, schema_import: str = 'gn2pg_import', querystring: dict = <factory>)

Bases: object

Database connection settings

host: str
name: str
password: str
port: int = 5432
querystring: dict
schema_import: str = 'gn2pg_import'
user: str
class gn2pg.check_conf.Gn2PgConf(file: str)

Bases: object

Read config file and expose list of sources configuration

secure_dict(source_name) → Dict

Secure sensitive data for logging

property source_list: Dict[str, Any]

Return list of site configurations.

property version: str

Return version.

exception gn2pg.check_conf.Gn2PgConfException

Bases: Exception

An exception occurred while loading parameters.

class gn2pg.check_conf.Gn2PgSourceConf(source: int, config: Dict[str, Any])

Bases: object

Source conf generator

property data_type: str

Return data type (eg. “synthese” or any other type you want), used trigger with conditions, if “synthese”, then insert data into “gn_synthese.synthese” table

Returns:

Data type

Return type:

str

property database: Db

Return database settings

Returns:

_description_

Return type:

Db

property enable: bool

Return flag to enable or not source

Returns:

True if source is enabled

Return type:

bool

property export_id: int

Return export id, used to access to export

Returns:

GeoNature export_id

Return type:

int

property id_application: int

Return GeoNature id_application, used to login (CRUVED)

Returns:

GeoNature id_application, default is 3

Return type:

str

property lru_maxsize: int

Return database import schema

Returns:

Database import schema

Return type:

int

property max_page_length: int

Page size limit in an API list request.

Returns:

Page size

Return type:

int

property max_requests: int

Return database import schema

Returns:

Database import schema

Return type:

int

property max_retry: int

Return database import schema

Returns:

Database import schema

Return type:

int

property name: str

Return source name

Returns:

Source name

Return type:

str

property nb_threads: int

Get the number of computing threads

Returns:

The number of computing threads

Return type:

int

property query_strings: dict

Return flag to enable or not source

Returns:

Querystrings dictionnary

Return type:

dict

property retry_delay: int

Return database import schema

Returns:

Database import schema

Return type:

int

property source: int

Return source list position, used to identify source configuration in config file.

Returns:

Return source list position

Return type:

int

property std_name: str

Return a standardized source name, used to tag data source in db

Returns:

standardized Source name

Return type:

str

property unavailable_delay: int

Return database import schema

Returns:

Database import schema

Return type:

int

property url: str

Return GeoNature URL, used to access to export

Returns:

GeoNature URL (https://…)

Return type:

str

property user_name: str

Return source GeoNature username to login into GeoNature foreign instance

Returns:

GeoNature user username

Return type:

str

property user_password: str

Return source GeoNature user password to login into GeoNature foreign instance

Returns:

GeoNature User password

Return type:

str

exception gn2pg.check_conf.IncorrectParameter

Bases: Gn2PgConfException

Incorrect or missing parameter.

exception gn2pg.check_conf.MissingConfigurationFile

Bases: Gn2PgConfException

Incorrect or missing parameter.

class gn2pg.check_conf.Source(name: str, user_name: str, user_password: str, url: str, export_id: int, data_type: str, id_application: int = 3, enable: bool = True, last_action_date: str | None = None, query_strings: dict = <factory>)

Bases: object

Source connection settings

data_type: str
enable: bool = True
export_id: int
id_application: int = 3
last_action_date: str | None = None
name: str
query_strings: dict
url: str
user_name: str
user_password: str
class gn2pg.check_conf.Tuning(max_page_length: int = 1000, max_retry: int = 5, max_requests: int = 0, retry_delay: int = 5, unavailable_delay: int = 600, lru_maxsize: int = 32, nb_threads: int = 1)

Bases: object

Tuning settings

lru_maxsize: int = 32
max_page_length: int = 1000
max_requests: int = 0
max_retry: int = 5
nb_threads: int = 1
retry_delay: int = 5
unavailable_delay: int = 600

gn2pg.download module

Methods to download from VisioNature and store to file.

Methods

  • download_taxo_groups - Download and store taxo groups

Properties

class gn2pg.download.Data(config, backend)

Bases: DownloadGn

Implement store from observations controler.

Methods - store - Download by page and store to json

class gn2pg.download.DownloadGn(config: Gn2PgSourceConf, api_instance: DataAPI, backend: StorePostgresql)

Bases: object

Top class, not for direct use. Provides internal and template methods.

delete(page: str, queue: Queue) → None

Delete (or not) data in DB from a page download

Parameters:

  • page (str) – url to download

  • queue (Queue) – gather the progress

download(page: str, queue: Queue) → None

Download a page and store the progress in the provided queue

Parameters:

  • page (str) – url to download

  • queue (Queue) – gather the progress

exit()

Final log on exit

launch_threads(nb_threads: int, func: Callable, pages: list, store=True) → None

Launch 1 + nb_threads threads to execute a function func on a list of pages

Parameters:

  • nb_threads (int) – number of threads to compute the function on the pages

  • func (Callable) – function that each thread will call

  • pages (list) – list of pages

  • store (bool) – if True, display Storing in logger

property name: str

Return the controler name.

process_progress(page: str) → dict

Compute the progress of the task

Parameters:

page (str) – url to download

Returns:

dict containing items, len_items, total_len

Return type:

dict (dict)

store() → None

Store data into Database

property transfer_errors: int

Return the number of HTTP errors during this session.

update(since: str | None = None, actions: list | None = None) → None

[summary]

Parameters:

  • since (str) – DateTime limit to update.

  • actions (list) – Actions list (Insert > I, Update > U, Delete > D)

property version: str

Return version.

exception gn2pg.download.DownloadGnException

Bases: Exception

An exception occurred while handling download or store.

exception gn2pg.download.NotImplementedException

Bases: DownloadGnException

Feature not implemented.

gn2pg.env module

Environment file

gn2pg.env.CONFDIR = PosixPath('/home/runner/.gn2pg')

Config system directory (~/.gn2pg/)

gn2pg.env.LOGDIR = PosixPath('/home/runner/.gn2pg/log')

Log system directory (subdir of CONFDIR)

gn2pg.helpers module

Main cli functions

gn2pg.helpers.edit_config(file_path: str) → None

Open config file in a text editor

gn2pg.helpers.full_download(cfg_ctrl)

Performs a full download of all sites and controlers, based on configuration file.

gn2pg.helpers.full_download_1source(ctrl, cfg)

Downloads from a single controler.

gn2pg.helpers.init(file: str) → None

Init config file from template

Parameters:

file (str) – The name of the configuration file to create.

gn2pg.helpers.manage_configs(action: str, file: str | None = None)

Manage config files from config directory (list, read, edit)

gn2pg.helpers.update(cfg_ctrl)

[summary]

Parameters:

cfg_ctrl ([type]) – [description]

gn2pg.helpers.update_1source(ctrl, cfg)

[summary]

Parameters:

  • ctrl ([type]) – [description]

  • cfg ([type]) – [description]

gn2pg.main module

Program entry point

gn2pg.main.arguments(args)

Define and parse command arguments.

Parameters:

args ([str]) – command line parameters as list of strings

Returns:

command line parameters namespace

Return type:

argparse.Namespace

gn2pg.main.handle_config_commands(args) → None

Handle commands related to ‘config’.

gn2pg.main.handle_database_commands(args, cfg_ctrl) → None

Handle commands related to ‘config’.

gn2pg.main.handle_download_commands(args, cfg_ctrl) → bool

Handle commands that are not related to ‘manage’.

gn2pg.main.main(args) → None

Main entry point allowing external calls

Parameters:

args ([str]) – command line parameter list

gn2pg.main.run()

Zero-argument entry point for use with setuptools/distribute.

gn2pg.store_postgresql module

Methods to store data to Postgresql database.

class gn2pg.store_postgresql.DataItem(source: str, metadata: MetaData, conn: Any, elem: dict)

Bases: object

Properties of an observation, for writing to DB.

property conn: Any

Return db connection

Returns:

db connection

Return type:

str

property elem: dict

Return Single observation to process and store

Returns:

Observation

Return type:

str

property metadata: MetaData

Return SqlAlchemy metadata

Returns:

SqlAlchemy metadata

Return type:

str

property source: str

Return source name

Returns:

Source name

Return type:

str

class gn2pg.store_postgresql.PostgresqlUtils(config)

Bases: object

Provides create and delete Postgresql database method.

count_json_data()

Count observations stored in json table, by source and type.

Returns:

Count of observations by site and taxonomy.

Return type:

dict

create_json_tables() → None

Create all internal and jsonb tables.

custom_script(script: str = 'to_gnsynthese') → None

EXecute custom script on DB. eg.: triggers to populate local tables like GeoNature synthese

Parameters:

script (str, optional) – custom script path. Defaults to “to_gnsynthese”.

class gn2pg.store_postgresql.StorePostgresql(config)

Bases: object

Provides store to Postgresql database method.

delete_data(items: list, id_key_name: str = 'id_synthese', controler: str = 'data') → int

Delete observations stored in database.

Parameters:

  • items (list) – items to delete

  • id_key_name (str, optional) – id key name from source. Defaults to “id_synthese”.

  • controler (str, optional) – Name of API controler. Defaults to “data”.

Returns:

Count of items deleted.

Return type:

int

error_log(controler: str, item: dict, error: str, uuid: str = None, last_ts: datetime = datetime.datetime(2025, 6, 10, 13, 27, 56, 903602)) → None

Store errors in database

Parameters:

  • controler (str) – Controler name

  • item (dict) – Item

  • error (str) – SQLAlchemy Error

  • uuid (str, optional) – Data or metadata UUID. Defaults to None.

  • last_ts (datetime, optional) – [description]. Defaults to datetime.now().

import_get(controler: str) → str | None

Get last download timestamp from database.

Parameters:

controler (str) – Controler name

Returns:

Return last increment timestamp if exists

Return type:

Optional[str]

import_log(controler: str, values: dict | None = None)

Write download log entries to database.

Parameters:

  • controler (str) – Name of API controler.

  • values (dict, optional) – Field values. Defaults to None

store_1_data(controler: str, elem: dict, id_key_name: str = 'id_synthese', uuid_key_name: str = 'id_perm_sinp') → None

Store 1 item in db (using upsert statement)

Parameters:

  • controler (str) – Destionation table

  • elem (dict) – json data as dict

  • id_key_name (str, optional) – Data id in source database. Defaults to “id_synthese”.

  • uuid_key_name (str, optional) – data UUID. Defaults to “id_perm_sinp”.

store_1_metadata(controler: str, level: str, elem: dict, uuid_key_name: str = 'uuid')

Store 1 metadata item in db (using upsert statement)

store_data(controler: str, items: list[dict], id_key_name: str = 'id_synthese', uuid_key_name: str = 'id_perm_sinp') → Tuple[int, int, int]

Write items_dict to database.

Parameters:

  • controler (str) – Name of API controler.

  • items (list) – Data returned from API call.

  • id_key_name (str, optional) – id key name from source. Defaults to “id_synthese”.

  • uuid_key_name (str, optional) – uuid key name from source. Defaults to “id_perm_sinp”.

Returns:

items dict length

Return type:

int

property version

Return version.

exception gn2pg.store_postgresql.StorePostgresqlException

Bases: Exception

An exception occurred while handling download or store.

gn2pg.store_postgresql.db_url(config)

db connection settings

gn2pg.utils module

Some utils

class gn2pg.utils.BColors

Bases: object

Colors used for cli

color(color: str)

bash shell color code

class gn2pg.utils.XferStatus

Bases: object

List of transfer status

delete = 'delete'
failed = 'failed'
import_data = 'importing data'
init = 'init'
success = 'success'
gn2pg.utils.coalesce_in_dict(source: dict, key: str, default: Any) → Any

Coalesce function applyed on dict values

Parameters:

  • source (dict) – Source

  • key (str) – key

  • default (any) – Default value

Returns:

Any value

Return type:

any

gn2pg.utils.simplify(source: str) → str

Codify source name

Parameters:

source (str) – Original source name

Returns:

Codified source name

Return type:

str

Module contents

Outil d’import de données entre instances GeoNature (côté client)