sanmar_sdk.ftp

SanMar’s SFTP server, and readers for the data files on it.

Every reader is a generator: rows are parsed one at a time as the file is read, so even SanMar’s largest files never sit in memory. Readers take a path to a local file or an open text stream, so the same code parses a file streamed by SanMarFTP or one downloaded some other way.

Submodules

• sanmar_sdk.ftp.catalog
• sanmar_sdk.ftp.client
• sanmar_sdk.ftp.inventory
• sanmar_sdk.ftp.legacy
• sanmar_sdk.ftp.orders
• sanmar_sdk.ftp.pricing
• sanmar_sdk.ftp.readers

Classes

CatalogProduct	One style, color and size from SanMar_SDL_N.csv or SanMar_EPDD.csv.
ProductInformation	One style, color and size from a SanMarPI file.
ProductRecord	The columns every SanMar product file shares.
SanMarFTP	A connection to SanMar's SFTP server.
WarehouseInventory	One style, color and size at one warehouse.
SaleItem	One style, color and size on sale, from sanmar_saleItems.txt.
TextFileProduct	One style, color and size from sanmar_pdd.txt or Catalog.txt.
TextFileRecord	The columns SanMar's older product text files share.
HoldingLine	One line of SanMar's acknowledgement of an SFTP order.
OrderFiles	The two files that place a batch of orders: names and contents.
ShipmentStatus	One line of one box in SanMar's daily shipment status file.
CustomerPrice	The account's price for one style, color and size.
PriceChange	A changed price from sanmar_dpc.csv.

Functions

read_catalog(→ collections.abc.Iterator[CatalogProduct])	Stream SanMar_SDL_N.csv (or SanMar_SDL_DI.csv), one product at a time.
read_extended_catalog(...)	Stream SanMar_EPDD.csv, one product at a time, with total stock.
read_product_information(...)	Stream a SanMarPI bulk, delta, brand or category file, one product at a time.
read_active_products(...)	Stream sanmar_activeproductsexport.txt, one row at a time.
read_warehouse_inventory(...)	Stream sanmar_dip.txt (or sanmar_closeouts_dip.txt), one row at a time.
read_catalog_txt(...)	Stream Catalog.txt, one product at a time.
read_pdd(→ collections.abc.Iterator[TextFileProduct])	Stream sanmar_pdd.txt, one product at a time.
read_sale_items(→ collections.abc.Iterator[SaleItem])	Stream sanmar_saleItems.txt, one sale item at a time.
read_holding(→ collections.abc.Iterator[HoldingLine])	Stream a Holding file, one order line at a time.
read_shipment_status(...)	Stream a daily shipment status file, one line of one box at a time.
render_order_files(→ OrderFiles)	Build the CustInfo.txt and Details.txt files for a batch of orders.
render_release_file(→ tuple[str, str])	Build the file that releases some or all of a batch's orders, as (name, contents).
read_customer_prices(...)	Stream sanmar_dp.csv or sanmar_dpIncentive.csv, one row at a time.
read_price_changes(→ collections.abc.Iterator[PriceChange])	Stream sanmar_dpc.csv, one changed price at a time.

Package Contents

class sanmar_sdk.ftp.CatalogProduct(/, **data: Any)

Bases: ProductRecord

One style, color and size from SanMar_SDL_N.csv or SanMar_EPDD.csv.

quantity: int | None = None

Stock across all warehouses, capped by SanMar. Only the EPDD file has it.

category: str | None = None

The category. In SDL files this holds every category, separated by semicolons.

subcategory: str | None = None

suggested_price: sanmar_sdk.base.Price = None

msrp: sanmar_sdk.base.Price = None

companion_styles: str | None = None

front_model_image_url: str | None = None

back_model_image_url: str | None = None

front_flat_image_url: str | None = None

back_flat_image_url: str | None = None

product_measurements: str | None = None

pms_color: str | None = None

The Pantone (PMS) color. SanMar says it does not stand in for the product color.

gtin: str | None = None

decoration_spec_sheet: str | None = None

class sanmar_sdk.ftp.ProductInformation(/, **data: Any)

Bases: ProductRecord

One style, color and size from a SanMarPI file.

category: str | None = None

keywords: sanmar_sdk.base.CommaSeparated = None

piece_sale_price: sanmar_sdk.base.Price = None

case_sale_price: sanmar_sdk.base.Price = None

sale_start_date: sanmar_sdk.base.SanMarDate | None = None

sale_end_date: sanmar_sdk.base.SanMarDate | None = None

price_code: str | None = None

The suggested retail pricing code. A/P is 50%, B/Q 45%, C/R 40%, D/S 35%, E/T 30%.

front_flat: str | None = None

back_flat: str | None = None

front_model: str | None = None

back_model: str | None = None

side_model: str | None = None

three_q_model: str | None = None

class sanmar_sdk.ftp.ProductRecord(/, **data: Any)

Bases: sanmar_sdk.base.Record

The columns every SanMar product file shares.

unique_key: str

SanMar’s identifier for this style, color and size, and the PromoStandards part id.

style: str = None

catalog_color: str = None

The catalog (mainframe) color, which web services and orders expect.

color_name: str | None = None

The full color name, for display only.

size: str

inventory_key: int

size_index: int

title: str | None = None

description: str | None = None

brand: str | None = None

status: str | None = None

Coming Soon, New, Regular, or Discontinued. Coming Soon rows may be incomplete.

available_sizes: str | None = None

price_text: str | None = None

piece_weight: decimal.Decimal | None = None

Approximate weight per piece in pounds.

piece_price: sanmar_sdk.base.Price = None

The price per piece for five pieces or fewer of one style and color.

case_price: sanmar_sdk.base.Price = None

The price per piece when buying by the case.

case_size: int | None = None

map_price: sanmar_sdk.base.Price = None

The minimum advertised price.

brand_logo_image: str | None = None

thumbnail_image: str | None = None

color_swatch_image: str | None = None

product_image: str | None = None

spec_sheet: str | None = None

color_square_image: str | None = None

color_product_image: str | None = None

color_product_image_thumbnail: str | None = None

sanmar_sdk.ftp.read_catalog(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[CatalogProduct]

Stream SanMar_SDL_N.csv (or SanMar_SDL_DI.csv), one product at a time.

sanmar_sdk.ftp.read_extended_catalog(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[CatalogProduct]

Stream SanMar_EPDD.csv, one product at a time, with total stock.

sanmar_sdk.ftp.read_product_information(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[ProductInformation]

Stream a SanMarPI bulk, delta, brand or category file, one product at a time.

class sanmar_sdk.ftp.SanMarFTP(customer_number: int | str, password: str, *, host: str = HOST, port: int = PORT, host_key: str | None = None, known_hosts: pathlib.Path | None = None, timeout: float = 30.0)

A connection to SanMar’s SFTP server.

Use it as a context manager, and read files inside the with block:

with SanMarFTP(123456, "ftp-password", host_key="ssh-rsa AAAA...") as ftp:
    with ftp.catalog() as products:
        for product in products:
            ...

Parameters:

• customer_number – The SanMar customer number, which is the SFTP username.

• password – The FTP password SanMar issued. SanMar.com logins do not work here.

• host – SanMar’s SFTP server.

• port – SanMar’s SFTP server.

• host_key – The server’s public key, to trust it without a known_hosts entry.

• known_hosts – A known_hosts file to trust instead of the system’s.

• timeout – Seconds to wait for the connection.

host = 'ftp.sanmar.com'

port = 2200

__enter__() → Self

Connect.

__exit__(exc_type: type[BaseException] | None, exc: BaseException | None, traceback: types.TracebackType | None) → None

Disconnect.

connect() → None

Connect and log in, verifying the server’s host key.

close() → None

Disconnect. Safe to call more than once.

property sftp: paramiko.sftp_client.SFTPClient

The underlying paramiko SFTP session, for anything this class does not cover.

list_folder(folder: str = '') → list[str]

List a folder’s entries by name, sorted. Folder names match case-insensitively.

find(folder: str, pattern: str) → list[str]

List the files in a folder whose names match a shell-style pattern, ignoring case.

Names come back sorted by name. SanMar dates the brand and category files it writes on request as MM-DD-YYYY, which does not sort by date; newest() does:

ftp.find("SanMarPDD/SanMarPI", "Brand_OGIO_*.csv")

newest(folder: str, pattern: str) → str

Find the most recently dated file in a folder whose name matches a pattern.

Files are compared by the MM-DD-YYYY date SanMar puts in the names of the brand and category files it writes on request, then by name:

ftp.newest("SanMarPDD/SanMarPI", "Brand_OGIO_*.csv")

Raises:

NotFoundError – If no file matches.

resolve(path: str) → str

Find a path on the server, matching each part of it case-insensitively.

SanMar’s guides spell the same file SanMar_SDL_N.csv and Sanmar_SDL_N.csv. An exact match wins over a case-insensitive one.

Raises:

NotFoundError – If some part of the path does not exist; the error lists what does.

open(path: str, *, prefetch: bool = True) → collections.abc.Iterator[IO[bytes]]

Open a remote file for reading, as bytes.

prefetch requests the whole file ahead of reading, which is much faster for reading straight through; turn it off for random access.

open_text(path: str, *, member: str | None = None, encoding: str = 'utf-8-sig') → collections.abc.Iterator[TextIO]

Open a remote text file, or a text file inside a remote zip archive.

For a .zip path, member names the file inside it; it may be left out when the archive holds a single .csv or .txt file.

download(path: str, destination: pathlib.Path) → pathlib.Path

Copy a remote file to destination (a file, or a folder to put it in).

stream[R](path: str, reader: collections.abc.Callable[[TextIO], collections.abc.Iterator[R]], *, member: str | None = None, encoding: str = 'utf-8-sig') → collections.abc.Iterator[collections.abc.Iterator[R]]

Stream a remote file through one of this package’s readers.

The rows are read as the file downloads, and only while the with block is open:

with ftp.stream("SanMarPDD/sanmar_saleItems.txt", read_sale_items) as items:
    for item in items:
        ...

catalog() → contextlib.AbstractContextManager[collections.abc.Iterator[sanmar_sdk.ftp.catalog.CatalogProduct]]

Stream SanMar_SDL_N.csv, SanMar’s main product file.

extended_catalog() → contextlib.AbstractContextManager[collections.abc.Iterator[sanmar_sdk.ftp.catalog.CatalogProduct]]

Stream SanMar_EPDD.csv: the product file, with stock across all warehouses.

warehouse_inventory() → contextlib.AbstractContextManager[collections.abc.Iterator[sanmar_sdk.ftp.inventory.WarehouseInventory]]

Stream sanmar_dip.txt: stock by warehouse, with current prices, refreshed hourly.

product_information(file_name: str) → contextlib.AbstractContextManager[collections.abc.Iterator[sanmar_sdk.ftp.catalog.ProductInformation]]

Stream a file the product information web service wrote to SanMarPDD/SanMarPI.

upload_orders(batch: str, orders: collections.abc.Sequence[sanmar_sdk.orders.PurchaseOrder]) → sanmar_sdk.ftp.orders.OrderFiles

Upload a batch of orders, without releasing them for processing.

SanMar acknowledges the batch with a Holding file (see holding()) and holds the orders until release_orders() releases them, for up to two weeks. These are production orders; SanMar has no test environment for SFTP ordering.

release_orders(batch: str, po_numbers: collections.abc.Sequence[str], *, release_number: int = 1, delay: float = 5.0) → str

Release some or all of an uploaded batch’s orders for processing.

Each release of the same batch needs the next release_number. SanMar asks for a pause of several seconds between uploading a batch and releasing it; delay waits that long first. Returns the release file’s name.

holding(batch: str) → collections.abc.Iterator[collections.abc.Iterator[sanmar_sdk.ftp.orders.HoldingLine]]

Stream SanMar’s acknowledgement of an uploaded batch, once it exists.

SanMar writes it to Holding within about 15 minutes of the upload, and moves it to Done once the orders are processed. The file is the one whose name is the batch name followed by Holding, so batch 06-07-2022-1 never reads 06-07-2022-10Holding.txt.

Raises:

NotFoundError – If there is no Holding file for the batch yet.

class sanmar_sdk.ftp.WarehouseInventory(/, **data: Any)

Bases: sanmar_sdk.base.Record

One style, color and size at one warehouse.

inventory_key: int

size_index: int

style: str = None

catalog_color: str

The catalog (mainframe) color, which web services and orders expect.

size: str

warehouse: sanmar_sdk.common.WarehouseNumber = None

quantity: int

Stock at this warehouse, capped by SanMar.

unique_key: str | None = None

SanMar’s identifier for this style, color and size, and the PromoStandards part id.

piece_weight: decimal.Decimal | None = None

piece_price: sanmar_sdk.base.Price = None

case_price: sanmar_sdk.base.Price = None

case_size: int | None = None

piece_sale_price: sanmar_sdk.base.Price = None

case_sale_price: sanmar_sdk.base.Price = None

sale_start_date: sanmar_sdk.base.SanMarDate | None = None

sale_end_date: sanmar_sdk.base.SanMarDate | None = None

discontinued_code: str | None = None

Who discontinued the product, S for SanMar or M for the mill.

property discontinued: bool

Whether SanMar or the mill has discontinued the product.

sanmar_sdk.ftp.read_active_products(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[WarehouseInventory]

Stream sanmar_activeproductsexport.txt, one row at a time.

sanmar_sdk.ftp.read_warehouse_inventory(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[WarehouseInventory]

Stream sanmar_dip.txt (or sanmar_closeouts_dip.txt), one row at a time.

class sanmar_sdk.ftp.SaleItem(/, **data: Any)

Bases: TextFileRecord

One style, color and size on sale, from sanmar_saleItems.txt.

piece_sale_price: sanmar_sdk.base.Price = None

case_sale_price: sanmar_sdk.base.Price = None

sale_start_date: sanmar_sdk.base.SanMarDate | None = None

sale_end_date: sanmar_sdk.base.SanMarDate | None = None

class sanmar_sdk.ftp.TextFileProduct(/, **data: Any)

Bases: TextFileRecord

One style, color and size from sanmar_pdd.txt or Catalog.txt.

piece_price: sanmar_sdk.base.Price = None

case_price: sanmar_sdk.base.Price = None

class sanmar_sdk.ftp.TextFileRecord(/, **data: Any)

Bases: sanmar_sdk.base.Record

The columns SanMar’s older product text files share.

inventory_key: int

size_index: int = None

style: str = None

The style number, as printed in SanMar’s catalog.

catalog_color: str

The catalog (mainframe) color, which web services and orders expect.

size: str

brand: str | None = None

mill_style: str | None = None

The manufacturer’s own style number.

description: str | None = None

extended_description: str | None = None

case_quantity: int | None = None

weight: decimal.Decimal | None = None

Approximate weight per piece in pounds.

size_type: str | None = None

SanMar’s internal size code.

gtin: str | None = None

sanmar_sdk.ftp.read_catalog_txt(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[TextFileProduct]

Stream Catalog.txt, one product at a time.

sanmar_sdk.ftp.read_pdd(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[TextFileProduct]

Stream sanmar_pdd.txt, one product at a time.

sanmar_sdk.ftp.read_sale_items(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[SaleItem]

Stream sanmar_saleItems.txt, one sale item at a time.

class sanmar_sdk.ftp.HoldingLine(/, **data: Any)

Bases: sanmar_sdk.base.Record

One line of SanMar’s acknowledgement of an SFTP order.

po_number: str = None

style: str

catalog_color: str = None

size: str

quantity: int = None

warehouse: sanmar_sdk.common.WarehouseNumber = None

The warehouse SanMar will ship this line from.

available: bool = None

Whether the stock is there. If not, SanMar’s customer service calls about it.

class sanmar_sdk.ftp.OrderFiles

The two files that place a batch of orders: names and contents.

cust_info_name: str

cust_info: str

details_name: str

details: str

class sanmar_sdk.ftp.ShipmentStatus(/, **data: Any)

Bases: sanmar_sdk.base.Record

One line of one box in SanMar’s daily shipment status file.

po_number: str = None

sales_order_number: str = None

ship_date: sanmar_sdk.base.SanMarDate

style: str

catalog_color: str = None

size: str

quantity: int = None

ship_from: str | None = None

ship_to_name: str | None = None

attention: str | None = None

ship_to_address1: str | None = None

ship_to_address2: str | None = None

ship_to_city: str | None = None

ship_to_state: str | None = None

ship_to_zip: str | None = None

ship_to_country: str | None = None

sub_total: sanmar_sdk.base.Price = None

freight: sanmar_sdk.base.Price = None

handling_fee: sanmar_sdk.base.Price = None

invoice_total: sanmar_sdk.base.Price = None

tracking_number: str | None = None

total_cases: int | None = None

box_number: int | None = None

description: str | None = None

inventory_key: int | None = None

size_index: int | None = None

invoice_attention: str | None = None

license_plate: str | None = None

The license plate number on the box’s label, for the packing slip service.

sanmar_sdk.ftp.read_holding(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[HoldingLine]

Stream a Holding file, one order line at a time.

sanmar_sdk.ftp.read_shipment_status(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[ShipmentStatus]

Stream a daily shipment status file, one line of one box at a time.

sanmar_sdk.ftp.render_order_files(batch: str, orders: collections.abc.Sequence[sanmar_sdk.orders.PurchaseOrder]) → OrderFiles

Build the CustInfo.txt and Details.txt files for a batch of orders.

Parameters:

• batch – The batch name, unique for every batch ever sent. SanMar suggests the date and a running number for the day, such as 06-07-2022-1.

• orders – The purchase orders in the batch. Items must be named by SkuKey, and each order needs a ship-to email.

sanmar_sdk.ftp.render_release_file(batch: str, po_numbers: collections.abc.Sequence[str], release_number: int = 1) → tuple[str, str]

Build the file that releases some or all of a batch’s orders, as (name, contents).

Each release of a batch gets the next release_number, starting at 1. PO numbers are checked as a PurchaseOrder checks its own.

class sanmar_sdk.ftp.CustomerPrice(/, **data: Any)

Bases: sanmar_sdk.base.Record

The account’s price for one style, color and size.

unique_key: str

inventory_key: int

size_index: int

style: str = None

catalog_color: str

The catalog (mainframe) color, which web services and orders expect.

size: str

my_price: sanmar_sdk.base.Price

The account’s price, which is the case or sale price where there is no special pricing.

piece_weight: decimal.Decimal | None = None

piece_price: sanmar_sdk.base.Price = None

case_price: sanmar_sdk.base.Price = None

case_size: int | None = None

piece_sale_price: sanmar_sdk.base.Price = None

case_sale_price: sanmar_sdk.base.Price = None

sale_start_date: sanmar_sdk.base.SanMarDate | None = None

sale_end_date: sanmar_sdk.base.SanMarDate | None = None

discontinued_code: str | None = None

Who discontinued the product, S for SanMar or M for the mill.

class sanmar_sdk.ftp.PriceChange(/, **data: Any)

Bases: sanmar_sdk.base.Record

A changed price from sanmar_dpc.csv.

unique_key: str

my_price: sanmar_sdk.base.Price

sanmar_sdk.ftp.read_customer_prices(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[CustomerPrice]

Stream sanmar_dp.csv or sanmar_dpIncentive.csv, one row at a time.

sanmar_sdk.ftp.read_price_changes(source: sanmar_sdk.ftp.readers.Source, *, encoding: str = 'utf-8-sig') → collections.abc.Iterator[PriceChange]

Stream sanmar_dpc.csv, one changed price at a time.